diff --git a/src/config.rs b/src/config.rs index 68cb5096..ee31fffe 100644 --- a/src/config.rs +++ b/src/config.rs @@ -230,6 +230,14 @@ pub struct StorageConfig { /// Number of rows to delete per chunk during storage enforcement #[serde(default = "default_cleanup_chunk_size")] pub cleanup_chunk_size: usize, + /// Time window (hours) during which high-severity (err/crit/alert/emerg) logs + /// are protected from disk-pressure deletion. 0 = disable the err+ floor. + #[serde(default = "default_err_floor_window_hours")] + pub err_floor_window_hours: u64, + /// Maximum err+ rows protected per source IP within the floor window. Bounds + /// any single source's share of the protected set. 0 = disable the floor. + #[serde(default = "default_err_floor_per_source_cap")] + pub err_floor_per_source_cap: usize, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -463,10 +471,29 @@ fn default_recovery_db_size_mb() -> u64 { 900 } fn default_min_free_disk_mb() -> u64 { - 512 + // 0 = disabled. Whole-filesystem free space is an EXTERNAL condition cortex + // cannot resolve by deleting its OWN data — see syslog-mcp-w4hh. When this is + // non-zero the enforcement path treats low free disk as a write-block signal, + // NOT as a trigger to self-trim. Default 0 to stop the self-wipe out of the box. + 0 } fn default_recovery_free_disk_mb() -> u64 { - 768 + // MUST stay paired with default_min_free_disk_mb: validate_storage_config + // requires recovery_free_disk_mb == 0 when min_free_disk_mb == 0, so a fresh + // StorageConfig::default() would FAIL validation if this were non-zero. + 0 +} +fn default_err_floor_window_hours() -> u64 { + // err+ rows received within this window are protected from disk-pressure + // deletion. Time-windowed (not unbounded) so an unauthenticated source cannot + // pin the floor indefinitely with severity=err spam (syslog-mcp-w4hh W1). + 24 +} +fn default_err_floor_per_source_cap() -> usize { + // Maximum err+ rows protected per source IP within the window. Bounds how much + // of the protected set any single (attacker-controlled) source can occupy, so + // one hostile sender cannot monopolise the floor (syslog-mcp-w4hh W1). + 10_000 } fn default_cleanup_interval_secs() -> u64 { 60 @@ -537,6 +564,8 @@ impl Default for StorageConfig { recovery_free_disk_mb: default_recovery_free_disk_mb(), cleanup_interval_secs: default_cleanup_interval_secs(), cleanup_chunk_size: default_cleanup_chunk_size(), + err_floor_window_hours: default_err_floor_window_hours(), + err_floor_per_source_cap: default_err_floor_per_source_cap(), } } } @@ -713,6 +742,14 @@ impl Config { "CORTEX_CLEANUP_CHUNK_SIZE", &mut config.storage.cleanup_chunk_size, )?; + env_override_parse( + "CORTEX_ERR_FLOOR_WINDOW_HOURS", + &mut config.storage.err_floor_window_hours, + )?; + env_override_parse( + "CORTEX_ERR_FLOOR_PER_SOURCE_CAP", + &mut config.storage.err_floor_per_source_cap, + )?; // [mcp.auth] env overrides. env_override_auth_mode("CORTEX_AUTH_MODE", &mut config.mcp.auth.mode)?; @@ -1328,6 +1365,20 @@ fn validate_storage_config(storage: &StorageConfig) -> anyhow::Result<()> { )); } + // err+ retention floor (syslog-mcp-w4hh). The floor is dimensioned in + // (time window × per-source row count), NOT bytes, so there is no + // meaningful "floor < max_db_size_mb" byte comparison. The coherent + // invariant is that the floor must not be self-contradictory: if a window + // is configured, the per-source cap must be > 0, otherwise the floor would + // protect a non-empty time window yet retain zero rows from it — a silent + // footgun that re-enables the err+ self-wipe the floor exists to prevent. + if storage.err_floor_window_hours > 0 && storage.err_floor_per_source_cap == 0 { + return Err(anyhow::anyhow!( + "err_floor_per_source_cap must be > 0 when err_floor_window_hours is set \ + (a window with a zero per-source cap protects no err+ rows)" + )); + } + Ok(()) } @@ -1378,6 +1429,8 @@ impl StorageConfig { recovery_free_disk_mb: 0, cleanup_interval_secs: 60, cleanup_chunk_size: 1, + err_floor_window_hours: default_err_floor_window_hours(), + err_floor_per_source_cap: default_err_floor_per_source_cap(), } } } diff --git a/src/config_tests.rs b/src/config_tests.rs index da17c9e3..b2df590c 100644 --- a/src/config_tests.rs +++ b/src/config_tests.rs @@ -316,8 +316,11 @@ fn defaults_include_storage_budget_settings() { let cfg = Config::default(); assert_eq!(cfg.storage.max_db_size_mb, 1024); assert_eq!(cfg.storage.recovery_db_size_mb, 900); - assert_eq!(cfg.storage.min_free_disk_mb, 512); - assert_eq!(cfg.storage.recovery_free_disk_mb, 768); + // syslog-mcp-w4hh: free-disk guardrail defaults to 0 (disabled) so cortex + // does not self-wipe to chase external whole-filesystem pressure. The two + // free-disk fields MUST default to 0 together to pass validate_storage_config. + assert_eq!(cfg.storage.min_free_disk_mb, 0); + assert_eq!(cfg.storage.recovery_free_disk_mb, 0); assert_eq!(cfg.storage.cleanup_interval_secs, 60); } @@ -451,9 +454,10 @@ fn docker_ingest_requires_hosts_when_enabled() { config.hosts.clear(); let err = validate_docker_ingest_config(&config).unwrap_err(); - assert!(err - .to_string() - .contains("docker_ingest.hosts must not be empty")); + assert!( + err.to_string() + .contains("docker_ingest.hosts must not be empty") + ); } #[test] @@ -478,9 +482,10 @@ fn docker_ingest_rejects_duplicate_host_names() { }; let err = validate_docker_ingest_config(&config).unwrap_err(); - assert!(err - .to_string() - .contains("duplicate docker_ingest host name")); + assert!( + err.to_string() + .contains("duplicate docker_ingest host name") + ); } #[test] @@ -1017,3 +1022,61 @@ fn repo_local_oauth_config_rejects_allowed_emails_until_enforced() { "wrong error: {err}" ); } + +// ---- syslog-mcp-w4hh: storage budget defaults + validation ---- + +/// The self-wipe stop-the-bleed: min_free_disk_mb defaults to 0 so cortex does +/// not treat external whole-FS pressure as a trigger to delete its own data. +#[test] +fn min_free_disk_mb_default_is_zero() { + let storage = StorageConfig::default(); + assert_eq!( + storage.min_free_disk_mb, 0, + "min_free_disk_mb must default to 0 (no external-pressure self-wipe)" + ); +} + +/// W2 MUST-FIX: StorageConfig::default() must PASS validate_storage_config. +/// validate_storage_config rejects recovery_free_disk_mb != 0 when +/// min_free_disk_mb == 0, so default_recovery_free_disk_mb must also be 0 — else +/// fresh deploys crash at startup. StorageConfig::for_test uses 0/0 and so cannot +/// catch this; this test asserts the real Default impl. +#[test] +fn default_storage_config_passes_validation() { + let storage = StorageConfig::default(); + assert_eq!( + storage.recovery_free_disk_mb, 0, + "recovery_free_disk_mb must default to 0 to pair with min_free_disk_mb=0" + ); + validate_storage_config(&storage) + .expect("StorageConfig::default() must pass validate_storage_config (W2)"); +} + +/// A TOML config with no [storage] overrides must also deserialize to defaults +/// that pass validation — guards the serde-default path, not just Default::default. +#[test] +fn default_toml_storage_config_passes_validation() { + #[derive(serde::Deserialize)] + struct Wrapper { + #[serde(default)] + storage: StorageConfig, + } + let parsed: Wrapper = toml::from_str("").expect("empty config must deserialize"); + validate_storage_config(&parsed.storage) + .expect("default-deserialized StorageConfig must pass validation (W2)"); +} + +/// The err+ floor invariant: a window with a zero per-source cap is rejected. +#[test] +fn err_floor_window_with_zero_cap_is_rejected() { + let storage = StorageConfig { + err_floor_window_hours: 24, + err_floor_per_source_cap: 0, + ..StorageConfig::default() + }; + let err = validate_storage_config(&storage).unwrap_err(); + assert!( + err.to_string().contains("err_floor_per_source_cap"), + "wrong error: {err}" + ); +} diff --git a/src/db.rs b/src/db.rs index e982a324..7186e62a 100644 --- a/src/db.rs +++ b/src/db.rs @@ -27,8 +27,9 @@ pub use ingest::insert_logs_batch; pub(crate) use ingest::insert_logs_batch_in_tx; pub use maintenance::{ db_full_vacuum, db_incremental_vacuum, db_integrity_check, db_wal_checkpoint, - enforce_storage_budget, exceeds_trigger, get_storage_metrics, physical_size_bytes, - purge_by_tag_window, purge_old_heartbeats, purge_old_logs, DiskSpaceProbe, + enforce_storage_budget, enforce_storage_budget_with_state, exceeds_trigger, + get_storage_metrics, physical_size_bytes, purge_by_tag_window, purge_old_heartbeats, + purge_old_logs, DiskSpaceProbe, SystemDiskSpaceProbe, }; pub(crate) use maintenance::{db_pragma_i64, db_pragma_string, PragmaName}; pub use models::{ diff --git a/src/db/maintenance.rs b/src/db/maintenance.rs index a9590169..727d6907 100644 --- a/src/db/maintenance.rs +++ b/src/db/maintenance.rs @@ -13,7 +13,7 @@ pub trait DiskSpaceProbe { fn free_bytes(&self, path: &Path) -> Result; } -struct SystemDiskSpaceProbe; +pub struct SystemDiskSpaceProbe; impl DiskSpaceProbe for SystemDiskSpaceProbe { fn free_bytes(&self, path: &Path) -> Result { @@ -138,6 +138,32 @@ pub fn enforce_storage_budget_with_probe( pool: &DbPool, config: &StorageConfig, probe: &impl DiskSpaceProbe, +) -> Result { + // No prior write-block state — used by the initial enforcement call at + // startup, before any tick has run. Hysteresis is a no-op on the first call. + enforce_storage_budget_with_state(pool, config, probe, false) +} + +/// Storage-budget enforcement with the previous tick's `write_blocked` state +/// threaded in for hysteresis on the EXTERNAL disk-pressure path. +/// +/// Two INDEPENDENT policies (syslog-mcp-w4hh): +/// - **DB-size (self-trim):** `max_db_size_mb` measures cortex's OWN logical +/// bytes. Cortex can resolve this by trimming its own oldest data, so it +/// loops `delete_oldest_*_chunk` down to `recovery_db_size_mb` — UNLESS doing +/// so would breach the err+ retention floor, in which case it stops and sets +/// `write_blocked` rather than wiping irreplaceable high-severity history. +/// - **Free-disk (external pressure):** `min_free_disk_mb` measures the WHOLE +/// filesystem (statvfs). A neighbour process filling the shared volume is not +/// something cortex can fix by deleting its own rows, so it NEVER deletes for +/// this trigger — it sets `write_blocked` and relies on ingest back-pressure +/// (receiver/writer.rs) until free disk recovers. Hysteresis: block engages at +/// `min_free_disk_mb`, clears only at `recovery_free_disk_mb`. +pub fn enforce_storage_budget_with_state( + pool: &DbPool, + config: &StorageConfig, + probe: &impl DiskSpaceProbe, + prev_write_blocked: bool, ) -> Result { let recovery = recovery_targets(config); let mut deleted_rows = 0usize; @@ -165,15 +191,31 @@ pub fn enforce_storage_budget_with_probe( }); } - // Only enter the cleanup loop if we have actually exceeded a trigger. - if exceeds_trigger(&metrics, config) { - while !within_recovery(&metrics, &recovery, config) { + // EXTERNAL disk-pressure decision (no deletion). Evaluated with hysteresis + // from the previous tick's state so the block engages at `min_free_disk_mb` + // and clears only once free disk has climbed back to `recovery_free_disk_mb`. + // The self-trim loop below runs INDEPENDENTLY of this — both can be active in + // the same tick (DB over its cap AND the filesystem low on free space). + let mut disk_write_blocked = disk_pressure_write_blocked(&metrics, config, prev_write_blocked); + if disk_write_blocked { + tracing::warn!( + free_disk_bytes = ?metrics.free_disk_bytes, + min_free_disk_mb = config.min_free_disk_mb, + recovery_free_disk_mb = config.recovery_free_disk_mb, + "Free-disk pressure detected — blocking writes WITHOUT deleting own data \ + (external whole-filesystem condition; cortex cannot resolve it by self-trim)" + ); + } + + // SELF-TRIM loop: only the DB-size trigger drives deletion. Its recovery exit + // is `logical <= recovery_db_size_mb` (the free-disk arm never gates it). + if db_size_exceeds_trigger(&metrics, config) { + while !db_size_within_recovery(&metrics, &recovery, config) { tracing::warn!( logical_db_size_bytes = metrics.logical_db_size_bytes, physical_db_size_bytes = metrics.physical_db_size_bytes, - free_disk_bytes = ?metrics.free_disk_bytes, deleted_rows, - "Storage budget exceeded trigger — deleting oldest telemetry chunk" + "DB-size budget exceeded — self-trimming oldest telemetry chunk" ); let deleted_orphan_children = delete_orphan_heartbeat_children(pool)?; @@ -198,7 +240,8 @@ pub fn enforce_storage_budget_with_probe( } } Some(TelemetrySource::Logs) => { - let deleted = delete_oldest_logs_chunk(pool, config.cleanup_chunk_size)?; + let deleted = + delete_oldest_logs_chunk(pool, config.cleanup_chunk_size, config)?; DeletedTelemetryChunk { deleted_rows: deleted.deleted_rows, log_hostnames: deleted.hostnames, @@ -211,21 +254,52 @@ pub fn enforce_storage_budget_with_probe( source: TelemetrySource::Logs, }, }; + // Floor-protection fallthrough: if the OLDEST source was logs but the + // chunk was fully err+-floor-protected (0 deleted), deletable heartbeats + // may still exist (they are simply newer than the protected logs, so + // `oldest_telemetry_source` picked logs). Trim a heartbeat chunk before + // concluding nothing is deletable — otherwise a DB-size breach would + // prematurely block writes while reclaimable heartbeat space remains. + let deleted = if deleted.deleted_rows == 0 && deleted.source == TelemetrySource::Logs { + let hb = delete_oldest_heartbeats_chunk(pool, config.cleanup_chunk_size)?; + if hb > 0 { + tracing::info!( + deleted_rows = hb, + "Oldest logs were floor-protected; trimmed heartbeat chunk instead" + ); + } + DeletedTelemetryChunk { + deleted_rows: hb, + log_hostnames: Vec::new(), + source: TelemetrySource::Heartbeats, + } + } else { + deleted + }; + if deleted.deleted_rows == 0 { + // Could not delete any more deletable rows. This is either an empty + // DB or — the case the err+ floor exists for — every remaining row + // is floor-protected AND no heartbeats remain to trim. Either way we + // stop trimming and BLOCK writes rather than wiping protected err+ + // history to chase the DB cap. metrics = get_storage_metrics_with_probe(pool, config, probe)?; - let write_blocked = exceeds_trigger(&metrics, config); + let still_over = db_size_exceeds_trigger(&metrics, config); tracing::warn!( logical_db_size_bytes = metrics.logical_db_size_bytes, free_disk_bytes = ?metrics.free_disk_bytes, deleted_rows, - write_blocked, - "Storage budget enforcement could not delete more rows" + db_size_still_over = still_over, + "Self-trim halted — no further deletable rows (err+ floor reached \ + or DB empty); blocking writes instead of deleting protected data" ); return Ok(StorageEnforcementOutcome { metrics, recovery, deleted_rows, - write_blocked, + // Block if EITHER the DB is still over cap with nothing left to + // safely trim, OR the external disk pressure was already latched. + write_blocked: still_over || disk_write_blocked, }); } @@ -238,13 +312,17 @@ pub fn enforce_storage_budget_with_probe( total_deleted_rows = deleted_rows, source = ?deleted.source, affected_hosts = deleted.log_hostnames.len(), - "Deleted oldest telemetry chunk for storage recovery" + "Self-trimmed oldest telemetry chunk for storage recovery" ); all_hosts.extend(deleted.log_hostnames); metrics = get_storage_metrics_with_probe(pool, config, probe)?; } } + // Re-evaluate disk pressure against fresh metrics after any self-trim above + // (self-trim frees real bytes, which can lift free disk back over recovery). + disk_write_blocked = disk_pressure_write_blocked(&metrics, config, prev_write_blocked); + if deleted_rows > 0 { // Reconcile hosts once after all chunks — avoids N×3 SQL round-trips // (one per chunk × 3 queries per hostname) competing with the batch writer. @@ -269,6 +347,7 @@ pub fn enforce_storage_budget_with_probe( logical_db_size_bytes = metrics.logical_db_size_bytes, physical_db_size_bytes = metrics.physical_db_size_bytes, free_disk_bytes = ?metrics.free_disk_bytes, + write_blocked = disk_write_blocked, "Storage budget enforcement completed" ); @@ -276,7 +355,9 @@ pub fn enforce_storage_budget_with_probe( metrics, recovery, deleted_rows, - write_blocked: false, + // The DB-size path resolves by self-trim and never blocks here; the only + // reason to block on a clean completion is unresolved EXTERNAL disk pressure. + write_blocked: disk_write_blocked, }) } @@ -595,56 +676,149 @@ fn oldest_telemetry_source(pool: &DbPool) -> Result> { }) } -fn delete_oldest_logs_chunk(pool: &DbPool, chunk_size: usize) -> Result { +/// Delete the oldest chunk of log rows for DB-size self-trim, honouring the +/// err+ retention FLOOR (syslog-mcp-w4hh). +/// +/// The floor protects, per source IP, the most-recent `err_floor_per_source_cap` +/// rows whose `severity IN ('err','crit','alert','emerg')` received within the +/// last `err_floor_window_hours`. Those rows are EXCLUDED from the deletable set, +/// so self-trim destroys low-value telemetry first and never wipes recent, +/// per-source-bounded high-severity history to chase the DB-size cap. +/// +/// Two security bounds (W1) make this safe against unauthenticated syslog: +/// - **time window** — only recent err+ is protected, so a hostile source +/// cannot pin the floor indefinitely with old severity=err spam; +/// - **per-source cap** — keyed on `source_ip` (the socket peer, which the +/// sender cannot freely vary per packet), NOT the payload `hostname` (which +/// is attacker-controlled), so no single source can monopolise the floor. +/// +/// Returning `deleted_rows == 0` while the DB is still over cap is the signal to +/// the caller that the floor (or an empty deletable set) has been reached; the +/// caller converts that to `write_blocked` instead of deleting protected rows. +fn delete_oldest_logs_chunk( + pool: &DbPool, + chunk_size: usize, + config: &StorageConfig, +) -> Result { let conn = pool.get()?; + // Build the protected-id CTE + the deletable selection. When the floor is + // disabled (window or cap == 0) we fall back to the original unfiltered + // oldest-first selection. + let floor_enabled = config.err_floor_window_hours > 0 && config.err_floor_per_source_cap > 0; + + // Window start as an RFC3339 string comparable to `received_at`. + // + // `received_at` is stored with MILLISECOND precision and a `Z` suffix (see + // `app::time::rfc3339_z`, the syslog/docker/OTLP ingest paths). We MUST format + // `window_start` the same way: a second-precision string like + // "...:27Z" sorts AFTER "...:27.680Z" lexicographically (because 'Z'=0x5A > + // '.'=0x2E), so a coarser format would silently protect the wrong rows. + // + // Overflow handling: `err_floor_window_hours` is a u64 and `TimeDelta` is + // i64-hours-bounded. A pathological value would overflow the conversion or the + // subtraction. The old code mapped that to `None`, which downstream collapsed + // to `""` — and `received_at >= ""` is always true, so EVERY err+ row would be + // protected, defeating the trim entirely. Fail fast instead. + let window_start = if floor_enabled { + let hours = i64::try_from(config.err_floor_window_hours).map_err(|_| { + anyhow::anyhow!( + "err_floor_window_hours ({}) is too large to represent as a time delta", + config.err_floor_window_hours + ) + })?; + let delta = chrono::TimeDelta::try_hours(hours).ok_or_else(|| { + anyhow::anyhow!( + "err_floor_window_hours ({hours}) overflows the supported time-delta range" + ) + })?; + let start = Utc::now().checked_sub_signed(delta).ok_or_else(|| { + anyhow::anyhow!( + "err_floor_window_hours ({hours}) underflows the representable timestamp range" + ) + })?; + Some(start.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)) + } else { + None + }; + + // Common deletable-id selection. `source_ip` is stored as `ip:port`; we + // PARTITION on the IP portion only (strip the ephemeral port) so all packets + // from one peer share a single per-source budget. Window functions require + // SQLite >= 3.25 (rusqlite `bundled` ships 3.4x). + let deletable_select = if floor_enabled { + "SELECT id FROM logs \ + WHERE id NOT IN ( \ + SELECT id FROM ( \ + SELECT id, ROW_NUMBER() OVER ( \ + PARTITION BY substr(source_ip, 1, \ + CASE WHEN instr(source_ip, ':') > 0 \ + THEN instr(source_ip, ':') - 1 \ + ELSE length(source_ip) END) \ + ORDER BY received_at DESC, id DESC \ + ) AS rn \ + FROM logs \ + WHERE severity IN ('err','crit','alert','emerg') \ + AND received_at >= :window_start \ + ) WHERE rn <= :cap \ + ) \ + ORDER BY received_at ASC, id ASC LIMIT :chunk" + } else { + "SELECT id FROM logs ORDER BY received_at ASC, id ASC LIMIT :chunk" + }; + // Collect distinct hostnames from the chunk we're about to delete. - // Use a subquery instead of a dynamic IN-list to avoid SQLite expression - // depth limit (default 1000) at large chunk sizes. let hostnames: Vec = { - let mut stmt = conn.prepare( - "SELECT DISTINCT hostname FROM logs \ - WHERE id IN (SELECT id FROM logs ORDER BY received_at ASC, id ASC LIMIT ?1)", - )?; - let result = stmt - .query_map([chunk_size as i64], |row| row.get(0))? - .collect::>>()?; - result + let sql = format!("SELECT DISTINCT hostname FROM logs WHERE id IN ({deletable_select})"); + let mut stmt = conn.prepare(&sql)?; + let rows = if floor_enabled { + stmt.query_map( + rusqlite::named_params! { + ":window_start": window_start.as_deref().unwrap_or(""), + ":cap": config.err_floor_per_source_cap as i64, + ":chunk": chunk_size as i64, + }, + |row| row.get(0), + )? + .collect::>>()? + } else { + stmt.query_map( + rusqlite::named_params! { ":chunk": chunk_size as i64 }, + |row| row.get(0), + )? + .collect::>>()? + }; + rows }; - // Pre-flight: count high-severity rows in the chunk so we can warn the - // operator that disk-pressure cleanup is overriding the time-based - // retention exemption for err+ logs. - let high_severity_count: i64 = conn.query_row( - "SELECT COUNT(*) FROM logs \ - WHERE id IN (SELECT id FROM logs ORDER BY received_at ASC, id ASC LIMIT ?1) \ - AND severity IN ('err', 'crit', 'alert', 'emerg')", - [chunk_size as i64], - |row| row.get(0), - )?; - if high_severity_count > 0 { - tracing::warn!( - high_severity_count, - chunk_size, - "Storage enforcement deleting high-severity rows — \ - disk pressure overrides time-based retention exemption" - ); - } - - // Delete the oldest chunk using a subquery — O(1) SQL string size regardless - // of chunk_size, no expression depth issues. + // Delete the deletable chunk. When the floor is active, protected err+ rows + // are never in this set — so this path no longer overrides the err+ exemption. + // Serialize the DELETE behind the process-wide write lock (v1.1.3) so it + // never races other writers against SQLite's single write lock. + let delete_sql = format!("DELETE FROM logs WHERE id IN ({deletable_select})"); let _write_guard = crate::db::write_lock(); - let deleted_rows = conn.execute( - "DELETE FROM logs \ - WHERE id IN (SELECT id FROM logs ORDER BY received_at ASC, id ASC LIMIT ?1)", - [chunk_size as i64], - )?; + let deleted_rows = if floor_enabled { + conn.execute( + &delete_sql, + rusqlite::named_params! { + ":window_start": window_start.as_deref().unwrap_or(""), + ":cap": config.err_floor_per_source_cap as i64, + ":chunk": chunk_size as i64, + }, + )? + } else { + conn.execute( + &delete_sql, + rusqlite::named_params! { ":chunk": chunk_size as i64 }, + )? + }; tracing::debug!( deleted_rows, affected_hosts = hostnames.len(), chunk_size, - "Deleted oldest logs chunk" + floor_enabled, + "Deleted oldest deletable logs chunk (err+ floor honoured)" ); Ok(DeletedChunk { @@ -827,23 +1001,71 @@ fn recovery_targets(config: &StorageConfig) -> StorageRecovery { } } +/// Combined trigger: true if EITHER the DB-size cap or the free-disk floor is +/// breached. Retained for the read-only health/stats surfaces (queries.rs, +/// service.rs) that report whether writes are currently constrained — they want +/// the OR of both conditions. Enforcement itself uses the split helpers below so +/// the two pressures get distinct remediation. pub fn exceeds_trigger(metrics: &StorageMetrics, config: &StorageConfig) -> bool { - (config.max_db_size_mb > 0 - && metrics.logical_db_size_bytes > mb_to_bytes(config.max_db_size_mb)) - || (config.min_free_disk_mb > 0 - && metrics.free_disk_bytes.unwrap_or(0) < mb_to_bytes(config.min_free_disk_mb)) + db_size_exceeds_trigger(metrics, config) || disk_free_below_trigger(metrics, config) +} + +/// DB-size trigger: cortex's OWN logical bytes exceed `max_db_size_mb`. +/// Resolvable by self-trim. +fn db_size_exceeds_trigger(metrics: &StorageMetrics, config: &StorageConfig) -> bool { + config.max_db_size_mb > 0 && metrics.logical_db_size_bytes > mb_to_bytes(config.max_db_size_mb) } -fn within_recovery( +/// Free-disk trigger: whole-filesystem free space is below `min_free_disk_mb`. +/// EXTERNAL — never resolved by deleting cortex's own data. +fn disk_free_below_trigger(metrics: &StorageMetrics, config: &StorageConfig) -> bool { + // FAIL-CLOSED: when the free-disk guardrail is enabled (`min_free_disk_mb > 0`) + // but the statvfs probe failed (`free_disk_bytes == None`), treat free space as + // 0 (unknown == worst case) so the guardrail engages conservatively instead of + // silently disabling itself. With the guardrail disabled the function short- + // circuits on `> 0` and never inspects the probe at all. + config.min_free_disk_mb > 0 + && metrics.free_disk_bytes.unwrap_or(0) < mb_to_bytes(config.min_free_disk_mb) +} + +/// Self-trim recovery exit: the DB-size loop stops once logical size is at or +/// below `recovery_db_size_mb`. Deliberately ignores the free-disk arm so the +/// self-trim loop is NOT gated by an external condition it cannot fix. +fn db_size_within_recovery( metrics: &StorageMetrics, recovery: &StorageRecovery, config: &StorageConfig, ) -> bool { - let db_ok = config.max_db_size_mb == 0 - || metrics.logical_db_size_bytes <= recovery.logical_db_size_bytes; - let disk_ok = config.min_free_disk_mb == 0 - || metrics.free_disk_bytes.unwrap_or(0) >= recovery.free_disk_bytes.unwrap_or(0); - db_ok && disk_ok + config.max_db_size_mb == 0 || metrics.logical_db_size_bytes <= recovery.logical_db_size_bytes +} + +/// Hysteresis decision for the external free-disk write-block. +/// +/// - Below `min_free_disk_mb` → engage the block. +/// - At/above `recovery_free_disk_mb` → clear the block. +/// - In the (min, recovery) hysteresis band → keep whatever the previous tick +/// decided (`prev`). This needs prior state: the answer in the band is not a +/// pure function of current metrics, which is exactly why the block latches +/// instead of flapping at the trigger threshold. +fn disk_pressure_write_blocked( + metrics: &StorageMetrics, + config: &StorageConfig, + prev: bool, +) -> bool { + if config.min_free_disk_mb == 0 { + return false; + } + // FAIL-CLOSED: the guardrail is enabled here, so a failed statvfs probe + // (`None`) is treated as 0 free bytes (worst case) — the block engages rather + // than fails open. Mirrors `disk_free_below_trigger`. + let free = metrics.free_disk_bytes.unwrap_or(0); + if free < mb_to_bytes(config.min_free_disk_mb) { + true + } else if free >= mb_to_bytes(config.recovery_free_disk_mb) { + false + } else { + prev + } } fn mb_to_bytes(mb: u64) -> u64 { diff --git a/src/db/maintenance_tests.rs b/src/db/maintenance_tests.rs index 56c81d86..9d53e687 100644 --- a/src/db/maintenance_tests.rs +++ b/src/db/maintenance_tests.rs @@ -1,6 +1,6 @@ use super::*; use crate::config::StorageConfig; -use crate::db::{init_pool, insert_logs_batch, list_hosts, tail_logs, DbPool, LogBatchEntry}; +use crate::db::{DbPool, LogBatchEntry, init_pool, insert_logs_batch, list_hosts, tail_logs}; use anyhow::Result; use rusqlite::params; use std::path::Path; @@ -367,28 +367,292 @@ impl DiskSpaceProbe for FakeDiskSpaceProbe { } } +/// syslog-mcp-w4hh: low whole-filesystem free space is an EXTERNAL condition. +/// Cortex must NOT delete its own data to chase it — it blocks writes instead. #[test] -fn test_enforce_storage_budget_recovers_when_free_disk_threshold_is_breached() { +fn external_disk_pressure_does_not_delete() { let (pool, dir) = test_pool(); let entries = vec![ - make_entry("2026-01-01T00:00:01Z", "deleted-host", "info", "older"), - make_entry("2026-01-01T00:00:02Z", "surviving-host", "info", "newer"), + make_entry("2026-01-01T00:00:01Z", "host-a", "info", "older"), + make_entry("2026-01-01T00:00:02Z", "host-b", "info", "newer"), ]; insert_logs_batch(&pool, &entries).unwrap(); update_received_at(&pool, "older", "2026-01-01T00:00:00Z"); update_received_at(&pool, "newer", "2026-01-02T00:00:00Z"); + // DB-size limit disabled; only the free-disk floor is active. The DB itself + // is tiny, so the disk pressure is genuinely external. let mut config = test_storage_config(dir.path().join("test.db")); config.max_db_size_mb = 0; config.recovery_db_size_mb = 0; config.min_free_disk_mb = 512; config.recovery_free_disk_mb = 768; - let probe = FakeDiskSpaceProbe::new(vec![64 * 1_048_576, 900 * 1_048_576]); + // Probe reports a persistently low free-disk value (well below the trigger). + let probe = FakeDiskSpaceProbe::new(vec![64 * 1_048_576]); let outcome = enforce_storage_budget_with_probe(&pool, &config, &probe).unwrap(); - assert!(outcome.deleted_rows > 0); - assert!(outcome.metrics.free_disk_bytes.unwrap() >= outcome.recovery.free_disk_bytes.unwrap()); + assert_eq!( + outcome.deleted_rows, 0, + "must NOT delete own data under external disk pressure" + ); + assert!( + outcome.write_blocked, + "must block writes while free disk is below the floor" + ); + + let rows = tail_logs(&pool, None, None, None, None, 10).unwrap(); + assert_eq!(rows.len(), 2, "both rows must survive — nothing deleted"); + + // The disk_fill alert decision is made in runtime.rs from the same metrics; + // verify the evaluator fires Some(..) at this free-disk level (the alert). + let critical = config.min_free_disk_mb * 1_048_576; + let warn = config.recovery_free_disk_mb * 1_048_576; + let params = crate::notifications::rules::evaluate_disk_fill( + "test-host", + 64 * 1_048_576, + critical, + warn, + "[]", + ); + assert!( + params.is_some(), + "disk_fill alert must fire at this free-disk level" + ); +} + +/// syslog-mcp-w4hh: when the DB grows past max_db_size_mb but the only remaining +/// rows are floor-protected err+ (recent window + within per-source cap), self-trim +/// must STOP at the floor and convert to write_blocked rather than wiping err+. +#[test] +fn self_trim_respects_err_floor() { + let (pool, dir) = test_pool(); + let now = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string(); + // One large, deletable info row (oldest) + several recent err+ rows that the + // floor protects. The err rows are big enough that, even after the info row is + // trimmed, the DB stays over the recovery target. + let big_info = "info-junk-".repeat(120_000); + let big_err1 = "err-keep-1-".repeat(120_000); + let big_err2 = "err-keep-2-".repeat(120_000); + insert_logs_batch( + &pool, + &[ + make_entry("2026-01-01T00:00:01Z", "host-a", "info", &big_info), + make_entry("2026-01-01T00:00:02Z", "host-a", "err", &big_err1), + make_entry("2026-01-01T00:00:03Z", "host-a", "crit", &big_err2), + ], + ) + .unwrap(); + // info row is oldest; err rows are received "now" so they are inside the window. + update_received_at(&pool, &big_info, "2026-01-01T00:00:00Z"); + update_received_at(&pool, &big_err1, &now); + update_received_at(&pool, &big_err2, &now); + + let mut config = test_storage_config(dir.path().join("test.db")); + config.max_db_size_mb = 2; + config.recovery_db_size_mb = 1; // recovery target the err rows alone exceed + config.min_free_disk_mb = 0; + config.recovery_free_disk_mb = 0; + config.cleanup_chunk_size = 1; + config.err_floor_window_hours = 24; + config.err_floor_per_source_cap = 10_000; + + let outcome = enforce_storage_budget(&pool, &config).unwrap(); + + // err+ rows must survive — the floor protected them. + let rows = tail_logs(&pool, None, None, None, None, 10).unwrap(); + let messages: Vec<&str> = rows.iter().map(|r| r.message.as_str()).collect(); + assert!( + messages.contains(&big_err1.as_str()), + "err row must survive the floor" + ); + assert!( + messages.contains(&big_err2.as_str()), + "crit row must survive the floor" + ); + assert!( + !messages.contains(&big_info.as_str()), + "the deletable info row should have been trimmed" + ); + // DB is still over cap (err+ retained), so writes must be blocked rather than + // the err+ history wiped. + assert!( + outcome.write_blocked, + "must block writes once trim reaches the err+ floor while still over cap" + ); +} + +/// Helper: insert a log with an explicit source_ip (the socket peer), used by the +/// W1 bound tests below to exercise the per-source partition. +fn make_entry_from( + ts: &str, + host: &str, + severity: &str, + source_ip: &str, + msg: &str, +) -> LogBatchEntry { + let mut e = make_entry(ts, host, severity, msg); + e.source_ip = source_ip.to_string(); + e +} + +/// syslog-mcp-w4hh W1 (monopolization defense): the per-source cap BOUNDS how +/// much err+ a single source IP can keep in the protected set. With cap=2, only +/// the 2 newest err+ rows from one source survive self-trim; the rest are +/// deletable even though they are inside the time window and high severity. +#[test] +fn err_floor_per_source_cap_evicts_excess() { + let (pool, dir) = test_pool(); + let now = chrono::Utc::now(); + // Five large err rows from the SAME source IP, all recent, staggered by second. + let mut msgs = Vec::new(); + for i in 0..5 { + let ts = (now - chrono::TimeDelta::seconds(10 - i)) + .format("%Y-%m-%dT%H:%M:%SZ") + .to_string(); + let msg = format!("err-{i}-{}", "z".repeat(700_000)); + insert_logs_batch( + &pool, + &[make_entry_from(&ts, "host-a", "err", "10.0.0.5:5000", &msg)], + ) + .unwrap(); + update_received_at(&pool, &msg, &ts); + msgs.push(msg); + } + + let mut config = test_storage_config(dir.path().join("test.db")); + config.max_db_size_mb = 2; + config.recovery_db_size_mb = 1; + config.min_free_disk_mb = 0; + config.recovery_free_disk_mb = 0; + config.cleanup_chunk_size = 1; + config.err_floor_window_hours = 24; + config.err_floor_per_source_cap = 2; // only 2 protected per source IP + + enforce_storage_budget(&pool, &config).unwrap(); + + let rows = tail_logs(&pool, None, None, None, None, 10).unwrap(); + let surviving: Vec<&str> = rows.iter().map(|r| r.message.as_str()).collect(); + // The 2 NEWEST err rows (indices 3,4) must survive; older ones evicted. + assert!( + surviving.contains(&msgs[4].as_str()) && surviving.contains(&msgs[3].as_str()), + "the 2 newest err rows from the source must be protected" + ); + assert!( + surviving.len() <= 2, + "per-source cap=2 must bound the source's protected err+; got {} survivors", + surviving.len() + ); + assert!( + !surviving.contains(&msgs[0].as_str()), + "the oldest err row must be evicted beyond the cap (monopolization bound)" + ); +} + +/// syslog-mcp-w4hh W1 (unbounded-pin defense): the time window BOUNDS how far +/// back the floor protects. err+ rows received OUTSIDE the window are deletable +/// by self-trim, so a hostile source cannot pin the DB at max with stale err spam. +#[test] +fn err_floor_window_evicts_stale_err() { + let (pool, dir) = test_pool(); + let big_stale_err = "stale-err-".repeat(120_000); + let big_recent_err = "recent-err-".repeat(120_000); + insert_logs_batch( + &pool, + &[ + make_entry_from( + "2026-01-01T00:00:01Z", + "host-a", + "err", + "10.0.0.9:6000", + &big_stale_err, + ), + make_entry_from( + "2026-01-01T00:00:02Z", + "host-a", + "err", + "10.0.0.9:6000", + &big_recent_err, + ), + ], + ) + .unwrap(); + let now = chrono::Utc::now(); + // Stale err: 48h ago, well outside a 1h window → NOT protected → deletable. + let stale_ts = (now - chrono::TimeDelta::hours(48)) + .format("%Y-%m-%dT%H:%M:%SZ") + .to_string(); + let recent_ts = now.format("%Y-%m-%dT%H:%M:%SZ").to_string(); + update_received_at(&pool, &big_stale_err, &stale_ts); + update_received_at(&pool, &big_recent_err, &recent_ts); + + let mut config = test_storage_config(dir.path().join("test.db")); + config.max_db_size_mb = 2; + config.recovery_db_size_mb = 1; + config.min_free_disk_mb = 0; + config.recovery_free_disk_mb = 0; + config.cleanup_chunk_size = 1; + config.err_floor_window_hours = 1; // 1h window — stale err falls outside + config.err_floor_per_source_cap = 10_000; + + enforce_storage_budget(&pool, &config).unwrap(); + + let rows = tail_logs(&pool, None, None, None, None, 10).unwrap(); + let surviving: Vec<&str> = rows.iter().map(|r| r.message.as_str()).collect(); + assert!( + !surviving.contains(&big_stale_err.as_str()), + "stale err+ outside the window must be deletable (unbounded-pin bound)" + ); + assert!( + surviving.contains(&big_recent_err.as_str()), + "recent err+ inside the window must still be protected" + ); +} + +/// syslog-mcp-w4hh: hysteresis. The external disk-pressure block engages at +/// min_free_disk_mb and clears only at recovery_free_disk_mb — in the band between +/// them the prior state is carried forward (latch, no flap). +#[test] +fn disk_pressure_write_block_uses_hysteresis() { + let (pool, dir) = test_pool(); + insert_logs_batch( + &pool, + &[make_entry("2026-01-01T00:00:01Z", "host-a", "info", "x")], + ) + .unwrap(); + + let mut config = test_storage_config(dir.path().join("test.db")); + config.max_db_size_mb = 0; + config.recovery_db_size_mb = 0; + config.min_free_disk_mb = 512; + config.recovery_free_disk_mb = 768; + + // Below trigger (512MB): engages regardless of prior state. + let low = FakeDiskSpaceProbe::new(vec![100 * 1_048_576]); + let blocked = enforce_storage_budget_with_state(&pool, &config, &low, false).unwrap(); + assert!(blocked.write_blocked, "below min: must block"); + assert_eq!(blocked.deleted_rows, 0, "must not delete on disk pressure"); + + // In the hysteresis band (600MB, between 512 and 768): keep prior state. + let band = FakeDiskSpaceProbe::new(vec![600 * 1_048_576]); + let still_blocked = enforce_storage_budget_with_state(&pool, &config, &band, true).unwrap(); + assert!( + still_blocked.write_blocked, + "in band with prev=true: stay blocked (latch)" + ); + let stays_clear = enforce_storage_budget_with_state(&pool, &config, &band, false).unwrap(); + assert!( + !stays_clear.write_blocked, + "in band with prev=false: stay clear (no premature engage)" + ); + + // At/above recovery (800MB): clear regardless of prior state. + let high = FakeDiskSpaceProbe::new(vec![800 * 1_048_576]); + let cleared = enforce_storage_budget_with_state(&pool, &config, &high, true).unwrap(); + assert!( + !cleared.write_blocked, + "at recovery threshold: must clear even if prev=true" + ); } #[test] @@ -575,3 +839,253 @@ fn test_purge_by_tag_window_respects_cutoff_boundary() { let messages: Vec<&str> = remaining.iter().map(|r| r.message.as_str()).collect(); assert!(messages.contains(&"fresh"), "fresh row must survive"); } + +/// A disk-space probe that always fails (simulates a statvfs/ENOENT error). +/// `get_storage_metrics_with_probe` maps the `Err` to `free_disk_bytes == None`. +#[derive(Clone)] +struct FailingDiskSpaceProbe; + +impl DiskSpaceProbe for FailingDiskSpaceProbe { + fn free_bytes(&self, _path: &Path) -> Result { + anyhow::bail!("simulated statvfs probe failure") + } +} + +/// syslog-mcp-w4hh (review bug #1 — FAIL-CLOSED): when the free-disk guardrail is +/// ENABLED but the disk-space probe fails (`free_disk_bytes == None`), the guardrail +/// must engage conservatively (treat unknown free space as the worst case) instead +/// of failing open. Previously `unwrap_or(u64::MAX)` made a probe failure look like +/// infinite free space, so the block NEVER engaged — defeating the safety behavior. +#[test] +fn probe_failure_engages_write_block_does_not_fail_open() { + let (pool, dir) = test_pool(); + let entries = vec![ + make_entry("2026-01-01T00:00:01Z", "host-a", "info", "older"), + make_entry("2026-01-01T00:00:02Z", "host-b", "info", "newer"), + ]; + insert_logs_batch(&pool, &entries).unwrap(); + + // DB-size limit disabled; only the free-disk guardrail is active and ENABLED. + let mut config = test_storage_config(dir.path().join("test.db")); + config.max_db_size_mb = 0; + config.recovery_db_size_mb = 0; + config.min_free_disk_mb = 512; + config.recovery_free_disk_mb = 768; + + // The probe fails on every call → free_disk_bytes is None. + let probe = FailingDiskSpaceProbe; + let outcome = enforce_storage_budget_with_probe(&pool, &config, &probe).unwrap(); + + assert!( + outcome.metrics.free_disk_bytes.is_none(), + "probe failure must surface as None, not a fabricated value" + ); + assert!( + outcome.write_blocked, + "FAIL-CLOSED: an enabled free-disk guardrail must block writes when the \ + probe fails (unknown == worst case), not fail open" + ); + // External pressure must never delete cortex's own data. + assert_eq!( + outcome.deleted_rows, 0, + "probe-failure pressure is external — must not delete own data" + ); + let rows = tail_logs(&pool, None, None, None, None, 10).unwrap(); + assert_eq!(rows.len(), 2, "both rows must survive — nothing deleted"); +} + +/// syslog-mcp-w4hh (review bug #1, unit-level): with the guardrail DISABLED +/// (`min_free_disk_mb == 0`), a probe failure must NOT engage the block — the +/// fail-closed behavior is scoped to the case where the operator asked for the +/// guardrail. This pins both halves of the trigger/write-block decision. +#[test] +fn probe_failure_with_guardrail_disabled_does_not_block() { + let metrics = StorageMetrics { + logical_db_size_bytes: 0, + physical_db_size_bytes: 0, + free_disk_bytes: None, // probe failed + }; + let mut config = StorageConfig::for_test(std::path::PathBuf::from("/tmp/x.db")); + config.min_free_disk_mb = 0; + config.recovery_free_disk_mb = 0; + + assert!( + !super::disk_free_below_trigger(&metrics, &config), + "disabled guardrail must not trigger on a failed probe" + ); + assert!( + !super::disk_pressure_write_blocked(&metrics, &config, false), + "disabled guardrail must not write-block on a failed probe" + ); + + // And with the guardrail ENABLED, the same None must engage both. + config.min_free_disk_mb = 512; + config.recovery_free_disk_mb = 768; + assert!( + super::disk_free_below_trigger(&metrics, &config), + "enabled guardrail must treat a failed probe as below the floor" + ); + assert!( + super::disk_pressure_write_blocked(&metrics, &config, false), + "enabled guardrail must write-block on a failed probe" + ); +} + +/// syslog-mcp-w4hh (review bug #2 — TIMESTAMP FORMAT): `received_at` is stored with +/// MILLISECOND precision (e.g. "...:27.680Z"). The err+ floor `window_start` must be +/// formatted the same way. A second-precision cutoff like "...:27Z" sorts AFTER +/// "...:27.680Z" lexicographically ('Z'=0x5A > '.'=0x2E), so a row that is genuinely +/// inside the window would be judged outside it and lose protection. This test pins +/// a recent err row whose received_at carries fractional seconds and verifies it is +/// protected by the floor (deleted_rows comes only from the deletable info row). +#[test] +fn err_floor_window_matches_fractional_second_received_at() { + let (pool, dir) = test_pool(); + let big_info = "info-junk-".repeat(120_000); + let big_err = "err-keep-".repeat(120_000); + insert_logs_batch( + &pool, + &[ + make_entry("2026-01-01T00:00:01Z", "host-a", "info", &big_info), + make_entry("2026-01-01T00:00:02Z", "host-a", "err", &big_err), + ], + ) + .unwrap(); + + // Oldest, deletable info row. + update_received_at(&pool, &big_info, "2026-01-01T00:00:00Z"); + + // Place the protected err row's received_at in the SAME WHOLE SECOND as the + // floor cutoff, with a `.999` fractional part. The cutoff is computed inside the + // function as `Utc::now() - window_hours`; we mirror that here. This is the only + // arrangement that exercises bug #2: a row a few seconds inside the window never + // reaches the fractional position (the date/second fields differ), so it passes + // under BOTH formats and proves nothing. + // + // Discrimination at the boundary second "HH:MM:SS": + // - Correct (Millis) cutoff "HH:MM:SS.mmmZ" with mmm < 999 → row ".999Z" >= + // cutoff → PROTECTED (this is the fix). + // - Buggy (second) cutoff "HH:MM:SSZ" → comparing "...SS.999Z" vs "...SSZ", + // the char after "SS" is '.'(0x2E) < 'Z'(0x5A), so row < cutoff → NOT + // protected → deleted. The old code would wipe a row that is genuinely + // inside the window. + // + // Race guard: the function's internal `now` is microseconds after ours. If our + // captured `now` is within the last ~50ms of a second, `now - window` could land + // in a LATER whole second than ours, making the boundary second mismatch and + // the test flake even WITH the fix. The function captures its own `Utc::now()` + // only after several DB ops (this UPDATE, metrics PRAGMAs + statvfs, orphan + // sweep, two MIN() queries) — tens of ms later. So require OUR `now` to be + // EARLY in the second (sub_ms < 500); that 500ms headroom dwarfs the elapsed + // time before the function computes `window_start`, guaranteeing both `now`s + // share the same whole second. + let window_hours = 1i64; + let boundary_second = loop { + let now = chrono::Utc::now(); + let sub_ms = now.timestamp_subsec_millis(); + if sub_ms < 500 { + // The cutoff second is floor(now - window_hours) to the whole second. + let cutoff = now - chrono::TimeDelta::hours(window_hours); + break cutoff.format("%Y-%m-%dT%H:%M:%S").to_string(); + } + std::thread::sleep(std::time::Duration::from_millis(20)); + }; + let recent_ts = format!("{boundary_second}.999Z"); + assert!( + recent_ts.contains(".999Z"), + "test fixture must sit at the boundary second with a .999 fractional, got {recent_ts}" + ); + update_received_at(&pool, &big_err, &recent_ts); + + let mut config = test_storage_config(dir.path().join("test.db")); + config.max_db_size_mb = 2; + config.recovery_db_size_mb = 1; + config.min_free_disk_mb = 0; + config.recovery_free_disk_mb = 0; + config.cleanup_chunk_size = 1; + config.err_floor_window_hours = window_hours as u64; + config.err_floor_per_source_cap = 10_000; + + let outcome = enforce_storage_budget(&pool, &config).unwrap(); + + let rows = tail_logs(&pool, None, None, None, None, 10).unwrap(); + let messages: Vec<&str> = rows.iter().map(|r| r.message.as_str()).collect(); + assert!( + messages.contains(&big_err.as_str()), + "recent err+ with fractional-second received_at must be protected by the floor" + ); + assert!( + !messages.contains(&big_info.as_str()), + "the deletable info row should have been trimmed" + ); + assert!( + outcome.write_blocked, + "still over cap after floor protected the err row → writes blocked" + ); +} + +/// syslog-mcp-w4hh (review bug #3 — heartbeat fallthrough): during a DB-SIZE breach, +/// when the OLDEST telemetry is logs but that chunk is fully err+-floor-protected +/// (delete returns 0), deletable heartbeats may still remain (newer than the +/// protected logs). The self-trim loop must fall through to trimming heartbeats +/// before declaring write_blocked, rather than blocking prematurely. +#[test] +fn self_trim_falls_through_to_heartbeats_when_logs_floor_protected() { + let (pool, dir) = test_pool(); + // A large, recent, floor-protected err log is the OLDEST telemetry. Heartbeats + // are NEWER, so oldest_telemetry_source picks logs first — and that chunk is + // fully protected (0 deleted). Deletable heartbeats remain. + let big_err = "err-protected-".repeat(120_000); + let now = chrono::Utc::now(); + let err_ts = + (now - chrono::TimeDelta::minutes(10)).to_rfc3339_opts(chrono::SecondsFormat::Millis, true); + let hb_ts = now.to_rfc3339_opts(chrono::SecondsFormat::Millis, true); + + insert_logs_batch(&pool, &[make_entry(&err_ts, "host-a", "err", &big_err)]).unwrap(); + update_received_at(&pool, &big_err, &err_ts); + // Insert several heartbeats (newer than the err row) — these are deletable. + for i in 0..5 { + insert_heartbeat(&pool, &format!("hb-host-{i}"), &hb_ts); + } + + let mut config = test_storage_config(dir.path().join("test.db")); + config.max_db_size_mb = 1; + config.recovery_db_size_mb = 1; + config.min_free_disk_mb = 0; + config.recovery_free_disk_mb = 0; + config.cleanup_chunk_size = 1; + config.err_floor_window_hours = 24; // protects the err row + config.err_floor_per_source_cap = 10_000; + + let hb_before: i64 = { + let conn = pool.get().unwrap(); + conn.query_row("SELECT COUNT(*) FROM host_heartbeats", [], |r| r.get(0)) + .unwrap() + }; + assert_eq!(hb_before, 5); + + let outcome = enforce_storage_budget(&pool, &config).unwrap(); + + // The protected err log must survive. + let rows = tail_logs(&pool, None, None, None, None, 10).unwrap(); + let messages: Vec<&str> = rows.iter().map(|r| r.message.as_str()).collect(); + assert!( + messages.contains(&big_err.as_str()), + "floor-protected err row must survive" + ); + // At least one heartbeat must have been trimmed (the fallthrough engaged) + // rather than the loop blocking immediately on the 0-deleted log chunk. + let hb_after: i64 = { + let conn = pool.get().unwrap(); + conn.query_row("SELECT COUNT(*) FROM host_heartbeats", [], |r| r.get(0)) + .unwrap() + }; + assert!( + hb_after < hb_before, + "deletable heartbeats must be trimmed via fallthrough (before: {hb_before}, after: {hb_after})" + ); + assert!( + outcome.deleted_rows > 0, + "fallthrough must report the heartbeat rows it trimmed" + ); +} diff --git a/src/lib.rs b/src/lib.rs index 8736d29f..1deb0f3f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -103,6 +103,8 @@ pub mod testing { recovery_free_disk_mb: 0, cleanup_interval_secs: 60, cleanup_chunk_size: 1, + err_floor_window_hours: 24, + err_floor_per_source_cap: 10_000, } } diff --git a/src/runtime.rs b/src/runtime.rs index 8de80552..b1d84637 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -652,13 +652,28 @@ impl RuntimeCore { }; let pool = Arc::clone(&storage_pool); let storage = storage_config.clone(); + // Carry the previous tick's write_blocked into enforcement so the + // external disk-pressure block latches with hysteresis (engage at + // min_free_disk_mb, clear only at recovery_free_disk_mb) rather than + // flapping at the trigger threshold (syslog-mcp-w4hh). + let prev_write_blocked = shared_storage_state + .lock() + .expect("storage state mutex poisoned") + .as_ref() + .map(|s| s.write_blocked) + .unwrap_or(false); tracing::debug!( cleanup_interval_secs = storage_config.cleanup_interval_secs, "Storage budget enforcement tick started" ); match tokio::task::spawn_blocking(move || { let _permit = permit; - let outcome = db::enforce_storage_budget(&pool, &storage)?; + let outcome = db::enforce_storage_budget_with_state( + &pool, + &storage, + &db::SystemDiskSpaceProbe, + prev_write_blocked, + )?; match db::db_wal_checkpoint(&pool, "passive") { Ok((busy, log_frames, checkpointed_frames)) => { if log_frames > 0 || busy != 0 { diff --git a/tests/enrich_pipeline.rs b/tests/enrich_pipeline.rs index c62aa3ce..90ca7752 100644 --- a/tests/enrich_pipeline.rs +++ b/tests/enrich_pipeline.rs @@ -25,6 +25,8 @@ fn make_pool() -> (DbPool, TempDir) { recovery_free_disk_mb: 0, cleanup_interval_secs: 60, cleanup_chunk_size: 1, + err_floor_window_hours: 24, + err_floor_per_source_cap: 10_000, }; let pool = init_pool(&config).expect("test db pool should init"); (pool, dir)