brokkr-broker::ws::eviction

Rust

Continuous retention enforcement for the agent telemetry tables (agent_k8s_events, agent_pod_logs).

Stance

Brokkr’s telemetry buffers exist for immediate operational support, not long-term retention. The hard ceiling is 6 hours. See the project_log_retention_stance memory and [[BROKKR-A-0008]]. Callers can construct a worker with a shorter retention but never longer — RetentionConfig::new clamps anything past 6h back down to 6h.

Eviction runs continuously on a tokio interval (default every 60s). created_at (server-side) is the eviction key — a misbehaving agent sending past-dated timestamps cannot keep ancient rows alive past the ceiling. The worker logs row counts at each tick for observability.

Structs

struct RetentionConfig

pub

Derives: Debug, Clone, Copy

Retention policy for the agent telemetry buffers.

Fields

Name	Type	Description
retention	Duration	Effective retention window — guaranteed <= HARD_RETENTION_CEILING.
tick_interval	Duration	How often the eviction worker runs.

Methods

new

pub

fn new(retention: Duration, tick_interval: Duration) -> Self

Construct a policy, clamping retention to the hard ceiling. Any caller asking for more than 6h silently gets 6h — this is the product invariant, not a configuration error.

Source

    pub fn new(retention: Duration, tick_interval: Duration) -> Self {
        let clamped = if retention > HARD_RETENTION_CEILING {
            HARD_RETENTION_CEILING
        } else {
            retention
        };
        Self {
            retention: clamped,
            tick_interval,
        }
    }

default_​policy

pub

fn default_policy() -> Self

Default policy: 6h retention, 60s tick.

Source

    pub fn default_policy() -> Self {
        Self::new(HARD_RETENTION_CEILING, DEFAULT_EVICTION_TICK)
    }

Functions

fn spawn

pub

fn spawn(dal: DAL, config: RetentionConfig) -> JoinHandle<()>

Spawn the continuous eviction worker. Returns the join handle so the caller can shut it down explicitly during tests; production drops it and the worker runs for the process lifetime.

Source

pub fn spawn(dal: DAL, config: RetentionConfig) -> JoinHandle<()> {
    tokio::spawn(async move {
        let mut interval = tokio::time::interval(config.tick_interval);
        // Skip the immediate fire — first eviction is one tick in.
        interval.tick().await;
        loop {
            interval.tick().await;
            run_once(&dal, config);
        }
    })
}

fn run_​once

pub

fn run_once(dal: &DAL, config: RetentionConfig)

Synchronous single eviction pass — exposed for tests so they can call it deterministically without waiting for the interval to fire.

Source

pub fn run_once(dal: &DAL, config: RetentionConfig) {
    metrics::ws_log_eviction_runs_total().inc();
    let cutoff = Utc::now() - chrono::Duration::from_std(config.retention).unwrap_or_default();
    match dal.agent_k8s_events().evict_older_than(cutoff) {
        Ok(n) => {
            if n > 0 {
                debug!(rows = n, "evicted agent_k8s_events older than retention");
                metrics::ws_telemetry_evicted_total("agent_k8s_events").inc_by(n as u64);
            }
        }
        Err(e) => warn!(error = %e, "agent_k8s_events eviction failed"),
    }
    match dal.agent_pod_logs().evict_older_than(cutoff) {
        Ok(n) => {
            if n > 0 {
                debug!(rows = n, "evicted agent_pod_logs older than retention");
                metrics::ws_telemetry_evicted_total("agent_pod_logs").inc_by(n as u64);
            }
        }
        Err(e) => warn!(error = %e, "agent_pod_logs eviction failed"),
    }
}

Constants

Name	Type	Value	Description
HARD_RETENTION_CEILING	Duration	Duration::from_secs(6 * 60 * 60)	Hard cap on retained telemetry — never configurable upward.
DEFAULT_EVICTION_TICK	Duration	Duration::from_secs(60)	Default eviction tick interval.