brokkr-broker::utils::background_tasks

Rust

Background tasks for the Brokkr Broker.

This module contains background tasks that run periodically to maintain system health and cleanup expired data.

Structs

struct DiagnosticCleanupConfig

pub

Configuration for diagnostic cleanup task.

Fields

Name	Type	Description
interval_seconds	u64	How often to run the cleanup (in seconds).
max_age_hours	i64	Maximum age for completed/expired diagnostics before deletion (in hours).

struct WorkOrderMaintenanceConfig

pub

Configuration for work order maintenance task.

Fields

Name	Type	Description
interval_seconds	u64	How often to run the maintenance (in seconds).

struct AgentMetricsRefreshConfig

pub

Configuration for the agent-metrics refresh task.

Fields

Name	Type	Description
interval_seconds	u64	How often to refresh the agent gauges (in seconds).

struct WebhookDeliveryConfig

pub

Configuration for webhook delivery worker.

Fields

Name	Type	Description
interval_seconds	u64	How often to poll for pending deliveries (in seconds).
batch_size	i64	Maximum number of deliveries to process per interval.

struct WebhookCleanupConfig

pub

Configuration for webhook cleanup task.

Fields

Name	Type	Description
interval_seconds	u64	How often to run the cleanup (in seconds).
retention_days	i64	Number of days to retain completed/dead deliveries.

struct AuditLogCleanupConfig

pub

Configuration for audit log cleanup task.

Fields

Name	Type	Description
interval_seconds	u64	How often to run the cleanup (in seconds).
retention_days	i64	Number of days to retain audit logs.

struct AgentEventsCleanupConfig

pub

Configuration for the agent-events cleanup task (BROKKR-T-0228).

Fields

Name	Type	Description
interval_seconds	u64	How often to run the cleanup (in seconds).
retention_days	i64	Number of days to retain agent events before hard-deletion.

Functions

fn start_​diagnostic_​cleanup_​task

pub

fn start_diagnostic_cleanup_task(
    dal: DAL,
    config: DiagnosticCleanupConfig,
    reloadable: Option<ReloadableConfig>,
)

Starts the diagnostic cleanup background task.

This task periodically:

1. Sweeps diagnostic requests past their expiry time into a terminal state — unclaimed pending requests become expired, abandoned claimed requests become failed (BROKKR-T-0300)
2. Deletes old completed/expired/failed diagnostic requests and their results

Parameters:

Name	Type	Description
dal	DAL	The Data Access Layer instance
config	DiagnosticCleanupConfig	Configuration for the cleanup task

Source

pub fn start_diagnostic_cleanup_task(
    dal: DAL,
    config: DiagnosticCleanupConfig,
    reloadable: Option<ReloadableConfig>,
) {
    info!(
        "Starting diagnostic cleanup task (interval: {}s, max_age: {}h)",
        config.interval_seconds, config.max_age_hours
    );

    tokio::spawn(async move {
        let mut current_interval = config.interval_seconds;
        let mut ticker = interval(Duration::from_secs(current_interval));

        loop {
            ticker.tick().await;

            // Re-read per tick so a reload takes effect without a restart
            // (BROKKR-T-0292). Falls back to the startup values when no
            // reloadable config was supplied (tests, and any caller that opts
            // out).
            let (want_interval, max_age_hours) = match &reloadable {
                Some(rc) => (
                    rc.diagnostic_cleanup_interval_seconds(),
                    rc.diagnostic_max_age_hours(),
                ),
                None => (config.interval_seconds, config.max_age_hours),
            };
            let config = DiagnosticCleanupConfig {
                interval_seconds: want_interval,
                max_age_hours,
            };
            if want_interval != current_interval {
                info!(
                    "Diagnostic cleanup interval changed {}s -> {}s (config reload)",
                    current_interval, want_interval
                );
                current_interval = want_interval;
                ticker = interval(Duration::from_secs(current_interval));
                // A fresh `interval` yields its first tick immediately; consume
                // it so the rebuild does not run this pass twice.
                ticker.tick().await;
            }

            // Sweep requests past their expiry time into a terminal state.
            // Unclaimed and abandoned requests are logged separately: the
            // former is routine, the latter means an agent took the work and
            // died holding it (BROKKR-T-0300).
            match dal.diagnostic_requests().expire_old_requests() {
                Ok(sweep) => {
                    if sweep.expired > 0 {
                        info!("Expired {} unclaimed diagnostic requests", sweep.expired);
                    }
                    if sweep.failed > 0 {
                        warn!(
                            "Failed {} diagnostic requests abandoned by their claiming agent",
                            sweep.failed
                        );
                    }
                }
                Err(e) => {
                    error!("Failed to expire diagnostic requests: {:?}", e);
                }
            }

            // Delete old completed/expired/failed requests (cascades to results)
            match dal
                .diagnostic_requests()
                .cleanup_old_requests(config.max_age_hours)
            {
                Ok(deleted) => {
                    if deleted > 0 {
                        info!(
                            "Cleaned up {} old diagnostic requests (age > {}h)",
                            deleted, config.max_age_hours
                        );
                    }
                }
                Err(e) => {
                    error!("Failed to cleanup old diagnostic requests: {:?}", e);
                }
            }
        }
    });
}

fn start_​work_​order_​maintenance_​task

pub

fn start_work_order_maintenance_task(
    dal: DAL,
    config: WorkOrderMaintenanceConfig,
)

Starts the work order maintenance background task.

This task periodically:

1. Moves RETRY_PENDING work orders back to PENDING when their backoff has elapsed
2. Reclaims stale CLAIMED work orders that have timed out

Parameters:

Name	Type	Description
dal	DAL	The Data Access Layer instance
config	WorkOrderMaintenanceConfig	Configuration for the maintenance task

Source

pub fn start_work_order_maintenance_task(dal: DAL, config: WorkOrderMaintenanceConfig) {
    info!(
        "Starting work order maintenance task (interval: {}s)",
        config.interval_seconds
    );

    tokio::spawn(async move {
        let mut ticker = interval(Duration::from_secs(config.interval_seconds));

        loop {
            ticker.tick().await;

            // Process RETRY_PENDING work orders whose backoff has elapsed
            match dal.work_orders().process_retry_pending() {
                Ok(count) => {
                    if count > 0 {
                        info!("Reset {} RETRY_PENDING work orders to PENDING", count);
                    }
                }
                Err(e) => {
                    error!("Failed to process retry pending work orders: {:?}", e);
                }
            }

            // Reclaim stale CLAIMED work orders
            match dal.work_orders().process_stale_claims() {
                Ok(count) => {
                    if count > 0 {
                        info!("Released {} stale claimed work orders", count);
                    }
                }
                Err(e) => {
                    error!("Failed to process stale claims: {:?}", e);
                }
            }
        }
    });
}

fn start_​agent_​metrics_​refresh_​task

pub

fn start_agent_metrics_refresh_task(
    dal: DAL,
    config: AgentMetricsRefreshConfig,
)

Starts the agent-metrics refresh background task (BROKKR-T-0226).

Periodically refreshes the brokkr_active_agents and brokkr_agent_heartbeat_age_seconds Prometheus gauges from the database so they stay correct independent of who calls GET /agents. Previously those gauges only updated inside the list_agents handler, so without scrape traffic on that path the heartbeat-age gauge would go stale and never grow.

Parameters:

Name	Type	Description
dal	DAL	The Data Access Layer instance
config	AgentMetricsRefreshConfig	Configuration for the refresh task

Source

pub fn start_agent_metrics_refresh_task(dal: DAL, config: AgentMetricsRefreshConfig) {
    info!(
        "Starting agent metrics refresh task (interval: {}s)",
        config.interval_seconds
    );

    tokio::spawn(async move {
        let mut ticker = interval(Duration::from_secs(config.interval_seconds));

        loop {
            ticker.tick().await;
            refresh_agent_metrics(&dal);
        }
    });
}

fn start_​webhook_​delivery_​task

pub

fn start_webhook_delivery_task(
    dal: DAL,
    config: WebhookDeliveryConfig,
    reloadable: Option<ReloadableConfig>,
)

Starts the webhook delivery worker background task.

This task periodically:

1. Releases expired acquired deliveries back to pending
2. Moves failed deliveries with elapsed backoff back to pending
3. Claims pending deliveries for broker (target_labels is NULL)
4. Attempts to deliver each via HTTP POST
5. Marks deliveries as success or failure (with retry scheduling)

Parameters:

Name	Type	Description
dal	DAL	The Data Access Layer instance
config	WebhookDeliveryConfig	Configuration for the delivery worker

Source

pub fn start_webhook_delivery_task(
    dal: DAL,
    config: WebhookDeliveryConfig,
    reloadable: Option<ReloadableConfig>,
) {
    info!(
        "Starting webhook delivery worker (interval: {}s, batch_size: {})",
        config.interval_seconds, config.batch_size
    );

    // No client-wide timeout: each request carries the timeout of its own
    // subscription (BROKKR-T-0288). A client-level timeout would silently cap
    // any subscription configured above it, which is exactly the broker-vs-agent
    // inconsistency this replaces.
    let client = reqwest::Client::builder()
        .build()
        .expect("Failed to create HTTP client");

    tokio::spawn(async move {
        let mut current_interval = config.interval_seconds;
        let mut ticker = interval(Duration::from_secs(current_interval));

        loop {
            ticker.tick().await;

            // Re-read per tick (BROKKR-T-0292). These were captured at spawn,
            // which is why the chart's `@hot-reload: true` annotations on them
            // were false.
            let (want_interval, batch_size) = match &reloadable {
                Some(rc) => (
                    rc.webhook_delivery_interval_seconds(),
                    rc.webhook_delivery_batch_size(),
                ),
                None => (config.interval_seconds, config.batch_size),
            };
            let config = WebhookDeliveryConfig {
                interval_seconds: want_interval,
                batch_size,
            };
            if want_interval != current_interval {
                info!(
                    "Webhook delivery interval changed {}s -> {}s (config reload)",
                    current_interval, want_interval
                );
                current_interval = want_interval;
                ticker = interval(Duration::from_secs(current_interval));
                ticker.tick().await;
            }

            // First, release any expired acquired deliveries
            match dal.webhook_deliveries().release_expired() {
                Ok(count) => {
                    if count > 0 {
                        debug!("Released {} expired webhook delivery claims", count);
                    }
                }
                Err(e) => {
                    error!("Failed to release expired webhook deliveries: {:?}", e);
                }
            }

            // Move failed deliveries back to pending if retry time has elapsed
            match dal.webhook_deliveries().process_retries() {
                Ok(count) => {
                    if count > 0 {
                        debug!(
                            "Moved {} webhook deliveries from failed to pending for retry",
                            count
                        );
                    }
                }
                Err(e) => {
                    error!("Failed to process webhook retries: {:?}", e);
                }
            }

            // Claim pending broker deliveries (target_labels is NULL)
            let deliveries = match dal
                .webhook_deliveries()
                .claim_for_broker(config.batch_size, None)
            {
                Ok(d) => d,
                Err(e) => {
                    error!("Failed to claim pending webhook deliveries: {:?}", e);
                    continue;
                }
            };

            if deliveries.is_empty() {
                continue;
            }

            debug!("Processing {} claimed webhook deliveries", deliveries.len());

            for delivery in deliveries {
                // Get the subscription to retrieve URL and auth header
                let subscription = match dal.webhook_subscriptions().get(delivery.subscription_id) {
                    Ok(Some(sub)) => sub,
                    Ok(None) => {
                        warn!(
                            "Subscription {} not found for delivery {}, marking as dead",
                            delivery.subscription_id, delivery.id
                        );
                        let _ = dal
                            .webhook_deliveries()
                            .mark_dead(delivery.id, "Subscription not found");
                        continue;
                    }
                    Err(e) => {
                        // The delivery is already claimed at this point, so
                        // `continue`ing here left the row `acquired` until its
                        // TTL lapsed, to be reclaimed and fail again forever,
                        // with `attempts` never incrementing (BROKKR-T-0307 --
                        // the broker-path sibling of T-0302/T-0304).
                        //
                        // `mark_dead` would be wrong, unlike the `Ok(None)` arm
                        // above: that means the subscription is genuinely gone,
                        // which is terminal, whereas this is a *database* read
                        // failure and is almost always transient. Discarding a
                        // subscriber's event because the broker briefly could
                        // not read its own table is the worse outcome.
                        //
                        // `mark_failed` is the right shape: it counts the
                        // attempt, releases the claim, schedules an exponential
                        // backoff, and reaches `dead` on its own if the failure
                        // persists. The ceiling is a fixed local constant
                        // because the subscription's own `max_retries` is
                        // exactly what could not be read -- and it bounds
                        // broker-side breakage, not subscriber tolerance.
                        let error = format!("Failed to load subscription: {e}");
                        error!(
                            "Failed to get subscription {} for delivery {}: {:?}",
                            delivery.subscription_id, delivery.id, e
                        );
                        if let Err(update_err) = dal.webhook_deliveries().mark_failed(
                            delivery.id,
                            &error,
                            SUBSCRIPTION_FETCH_MAX_ATTEMPTS,
                        ) {
                            // The same outage that broke the fetch can break
                            // this write. Nothing more to do here; the TTL
                            // release still applies, so the row is not lost --
                            // it simply retries as it did before this fix.
                            error!(
                                "Failed to record subscription-load failure for delivery {}: {:?}",
                                delivery.id, update_err
                            );
                        }
                        continue;
                    }
                };

                // Decrypt URL and auth header
                let url = match super::encryption::decrypt_string(&subscription.url_encrypted) {
                    Ok(u) => u,
                    Err(e) => {
                        error!(
                            "Failed to decrypt URL for subscription {}: {}",
                            subscription.id, e
                        );
                        crate::metrics::record_webhook_decrypt_failure("url", "broker");
                        let _ = dal
                            .webhook_deliveries()
                            .mark_dead(delivery.id, &format!("Failed to decrypt URL: {}", e));
                        continue;
                    }
                };

                let auth_header = subscription
                    .auth_header_encrypted
                    .as_ref()
                    .map(|encrypted| super::encryption::decrypt_string(encrypted))
                    .transpose();

                let auth_header = match auth_header {
                    Ok(h) => h,
                    Err(e) => {
                        error!(
                            "Failed to decrypt auth header for subscription {}: {}",
                            subscription.id, e
                        );
                        crate::metrics::record_webhook_decrypt_failure("auth_header", "broker");
                        let _ = dal.webhook_deliveries().mark_dead(
                            delivery.id,
                            &format!("Failed to decrypt auth header: {}", e),
                        );
                        continue;
                    }
                };

                // Attempt delivery, honoring this subscription's own timeout.
                // Clamped defensively: a row written before validation existed
                // could hold 0 or a negative, which would sign-extend into an
                // absurd timeout (mirrors `test_webhook`).
                let timeout = Duration::from_secs(subscription.timeout_seconds.max(1) as u64);
                let result = attempt_delivery(
                    &client,
                    &url,
                    auth_header.as_deref(),
                    &delivery.payload,
                    timeout,
                )
                .await;

                match result {
                    Ok(_) => match dal.webhook_deliveries().mark_success(delivery.id) {
                        Ok(_) => {
                            debug!(
                                "Webhook delivery {} succeeded for subscription {}",
                                delivery.id, subscription.id
                            );
                        }
                        Err(e) => {
                            error!(
                                "Failed to mark delivery {} as success: {:?}",
                                delivery.id, e
                            );
                        }
                    },
                    Err(failure) => {
                        let DeliveryFailure {
                            message: error,
                            status,
                            retryable,
                        } = failure;
                        // A non-retryable status (see `is_retryable_status`)
                        // will be returned identically on every attempt, so the
                        // delivery dies now instead of burning the whole
                        // attempt budget on it (BROKKR-T-0288).
                        let outcome = if retryable {
                            dal.webhook_deliveries().mark_failed(
                                delivery.id,
                                &error,
                                subscription.max_retries,
                            )
                        } else {
                            dal.webhook_deliveries().mark_dead(delivery.id, &error)
                        };
                        match outcome {
                            Ok(updated) => {
                                if updated.status == "dead" {
                                    warn!(
                                        "Webhook delivery {} dead after {} attempts: {}",
                                        delivery.id, updated.attempts, error
                                    );
                                    crate::utils::audit::log_action(
                                        ACTOR_TYPE_SYSTEM,
                                        None,
                                        ACTION_WEBHOOK_DELIVERY_FAILED,
                                        RESOURCE_TYPE_WEBHOOK,
                                        Some(delivery.subscription_id),
                                        Some(serde_json::json!({
                                            "delivery_id": delivery.id,
                                            "event_type": delivery.event_type,
                                            "attempts": updated.attempts,
                                            "error": error,
                                            "status_code": status,
                                            "retryable": retryable,
                                        })),
                                        None,
                                        None,
                                    );
                                } else {
                                    debug!(
                                        "Webhook delivery {} failed (attempt {}), will retry: {}",
                                        delivery.id, updated.attempts, error
                                    );
                                }
                            }
                            Err(e) => {
                                error!(
                                    "Failed to mark delivery {} as failed: {:?}",
                                    delivery.id, e
                                );
                            }
                        }
                    }
                }
            }
        }
    });
}

fn start_​webhook_​cleanup_​task

pub

fn start_webhook_cleanup_task(
    dal: DAL,
    config: WebhookCleanupConfig,
    reloadable: Option<ReloadableConfig>,
)

Starts the webhook cleanup background task.

This task periodically deletes old completed/dead deliveries based on the retention policy.

Parameters:

Name	Type	Description
dal	DAL	The Data Access Layer instance
config	WebhookCleanupConfig	Configuration for the cleanup task

Source

pub fn start_webhook_cleanup_task(
    dal: DAL,
    config: WebhookCleanupConfig,
    reloadable: Option<ReloadableConfig>,
) {
    info!(
        "Starting webhook cleanup task (interval: {}s, retention: {}d)",
        config.interval_seconds, config.retention_days
    );

    tokio::spawn(async move {
        let mut ticker = interval(Duration::from_secs(config.interval_seconds));

        loop {
            ticker.tick().await;

            // Retention is re-read per tick (BROKKR-T-0292). The interval is
            // not a dynamic key -- it is fixed at an hour by the caller -- so
            // there is no ticker to rebuild here.
            let retention_days = match &reloadable {
                Some(rc) => rc.webhook_cleanup_retention_days(),
                None => config.retention_days,
            };

            match dal.webhook_deliveries().cleanup_old(retention_days) {
                Ok(deleted) => {
                    if deleted > 0 {
                        info!(
                            "Cleaned up {} old webhook deliveries (age > {}d)",
                            deleted, retention_days
                        );
                    }
                }
                Err(e) => {
                    error!("Failed to cleanup old webhook deliveries: {:?}", e);
                }
            }
        }
    });
}

fn start_​audit_​log_​cleanup_​task

pub

fn start_audit_log_cleanup_task(dal: DAL, config: AuditLogCleanupConfig)

Starts the audit log cleanup background task.

This task periodically deletes old audit log entries based on the configured retention policy.

Parameters:

Name	Type	Description
dal	DAL	The Data Access Layer instance
config	AuditLogCleanupConfig	Configuration for the cleanup task

Source

pub fn start_audit_log_cleanup_task(dal: DAL, config: AuditLogCleanupConfig) {
    info!(
        "Starting audit log cleanup task (interval: {}s, retention: {}d)",
        config.interval_seconds, config.retention_days
    );

    tokio::spawn(async move {
        let mut ticker = interval(Duration::from_secs(config.interval_seconds));

        loop {
            ticker.tick().await;

            match dal.audit_logs().cleanup_old_logs(config.retention_days) {
                Ok(deleted) => {
                    if deleted > 0 {
                        info!(
                            "Cleaned up {} old audit logs (age > {}d)",
                            deleted, config.retention_days
                        );
                    }
                }
                Err(e) => {
                    error!("Failed to cleanup old audit logs: {:?}", e);
                }
            }
        }
    });
}

fn start_​agent_​events_​cleanup_​task

pub

fn start_agent_events_cleanup_task(
    dal: DAL,
    config: AgentEventsCleanupConfig,
)

Starts the agent-events cleanup background task (BROKKR-T-0228).

agent_events has no eviction of its own — rows are only soft-deleted on agent-delete cascade — so the table grows without bound at fleet scale. This task periodically hard-deletes events whose created_at is older than the configured retention window, keeping the table bounded while leaving the fleet activity feed (which serves the most recent N) intact.

The caller is responsible for only spawning this when retention_days > 0; a window of 0 disables eviction (the historical behavior).

Parameters:

Name	Type	Description
dal	DAL	The Data Access Layer instance
config	AgentEventsCleanupConfig	Configuration for the cleanup task

Source

pub fn start_agent_events_cleanup_task(dal: DAL, config: AgentEventsCleanupConfig) {
    info!(
        "Starting agent events cleanup task (interval: {}s, retention: {}d)",
        config.interval_seconds, config.retention_days
    );

    tokio::spawn(async move {
        let mut ticker = interval(Duration::from_secs(config.interval_seconds));

        loop {
            ticker.tick().await;

            let cutoff = chrono::Utc::now() - chrono::Duration::days(config.retention_days);
            match dal.agent_events().delete_older_than(cutoff) {
                Ok(deleted) => {
                    if deleted > 0 {
                        info!(
                            "Cleaned up {} old agent events (age > {}d)",
                            deleted, config.retention_days
                        );
                    }
                }
                Err(e) => {
                    error!("Failed to cleanup old agent events: {:?}", e);
                }
            }
        }
    });
}

fn start_​fleet_​sweep_​task

pub

fn start_fleet_sweep_task(
    dal: DAL,
    registry: std::sync::Arc<crate::ws::ConnectionRegistry>,
    fleet: std::sync::Arc<crate::ws::FleetBroadcaster>,
    interval_seconds: u64,
)

Starts the periodic fleet live-push sweep (the computed-signal half of the I-0028 hybrid trigger). Every interval_seconds it recomputes all fleet records and broadcasts a FleetUpdate for each agent whose computed signals changed. The first tick only seeds the baseline (no broadcast) so startup doesn’t blast the whole fleet to subscribers.

Source

pub fn start_fleet_sweep_task(
    dal: DAL,
    registry: std::sync::Arc<crate::ws::ConnectionRegistry>,
    fleet: std::sync::Arc<crate::ws::FleetBroadcaster>,
    interval_seconds: u64,
) {
    info!(
        "Starting fleet live-push sweep task (interval: {}s)",
        interval_seconds
    );
    tokio::spawn(async move {
        let mut ticker = interval(Duration::from_secs(interval_seconds));
        let mut last: std::collections::HashMap<uuid::Uuid, FleetComputedSignals> =
            std::collections::HashMap::new();
        let mut seeded = false;
        loop {
            ticker.tick().await;
            let records = match crate::api::v1::fleet::build_all_fleet_records(&dal, &registry) {
                Ok(records) => records,
                Err(e) => {
                    warn!("fleet sweep: failed to build fleet records: {:?}", e);
                    continue;
                }
            };
            let changed = select_changed_fleet_records(&mut last, &records);
            if !seeded {
                // First pass only populates the baseline — no broadcast.
                seeded = true;
                continue;
            }
            for record in changed {
                fleet.broadcast(brokkr_wire::WsMessage::FleetUpdate(record.to_wire()));
            }
        }
    });
}