Merge pull request #258 from trydirect/feature/agent-sweep

Feature/agent sweep

Author: vsilent

src/cli/stacker_client.rs

@@ -4096,6 +4096,21 @@ pub fn build_project_body(config: &StackerConfig) -> serde_json::Value {
         service_apps.push(service_to_app_json(svc, &network_ids));
     }
 
+    // `project_app.config_contract` is persisted by a separate server-side
+    // accessor. Include the full contract on each generated app so `stacker
+    // sync` does not silently discard the policy from stacker.yml.
+    if let Ok(config_contract) = serde_json::to_value(&config.config_contract) {
+        let has_services = config_contract
+            .get("services")
+            .and_then(serde_json::Value::as_object)
+            .is_some_and(|services| !services.is_empty());
+        if has_services {
+            for app in web_apps.iter_mut().chain(service_apps.iter_mut()) {
+                app["config_contract"] = config_contract.clone();
+            }
+        }
+    }
+
     serde_json::json!({
         "custom": {
             "custom_stack_code": stack_code,
@@ -5129,6 +5144,40 @@ mod tests {
         );
     }
 
+    #[test]
+    fn build_project_body_includes_config_contract_on_apps() {
+        let mut config = crate::cli::config_parser::ConfigBuilder::new()
+            .name("contract-project")
+            .app_image("nginx:1.27")
+            .build()
+            .expect("config should build");
+        config.config_contract = serde_json::from_value(serde_json::json!({
+            "services": {
+                "app": {
+                    "fields": {
+                        "JWT_SECRET": {
+                            "mutability": "generated",
+                            "type": "hex",
+                            "length": 32
+                        }
+                    }
+                }
+            }
+        }))
+        .expect("config contract should deserialize");
+
+        let body = build_project_body(&config);
+        let contract = &body["custom"]["web"][0]["config_contract"];
+        assert_eq!(
+            contract["services"]["app"]["fields"]["JWT_SECRET"]["mutability"],
+            "generated"
+        );
+        assert_eq!(
+            contract["services"]["app"]["fields"]["JWT_SECRET"]["type"],
+            "hex"
+        );
+    }
+
     #[test]
     fn pipe_instance_request_serializes_adapter_references() {
         let request = CreatePipeInstanceApiRequest {

src/db/agent.rs

@@ -239,6 +239,75 @@ pub async fn delete(pool: &PgPool, agent_id: Uuid) -> Result<(), String> {
     })
 }
 
+/// Delete agents whose deployment is gone (or soft-deleted) and that show no
+/// sign of life within the retention window.
+///
+/// "No sign of life" means both:
+///   - `last_heartbeat` is NULL or older than the window, AND
+///   - no `audit_log` row references the agent (by id or deployment_hash)
+///     within the same window.
+///
+/// The second condition protects agents that are alive but failing
+/// authentication — `last_heartbeat` only advances on successful `wait`/`report`,
+/// while `audit_log` captures `auth_failure` entries.
+#[tracing::instrument(name = "Sweep dead agents", skip(pool))]
+pub async fn sweep_dead(pool: &PgPool, retention_days: i32) -> Result<u64, String> {
+    let result = sqlx::query(
+        r#"
+        DELETE FROM agents a
+        WHERE NOT EXISTS (
+                SELECT 1 FROM deployment d
+                 WHERE d.deployment_hash = a.deployment_hash
+                   AND d.deleted IS NOT TRUE
+              )
+          AND (a.last_heartbeat IS NULL
+               OR a.last_heartbeat < NOW() - make_interval(days => $1))
+          AND NOT EXISTS (
+                SELECT 1 FROM audit_log l
+                 WHERE (l.agent_id = a.id OR l.deployment_hash = a.deployment_hash)
+                   AND l.created_at > NOW() - make_interval(days => $1)
+              )
+        "#,
+    )
+    .bind(retention_days)
+    .execute(pool)
+    .await
+    .map_err(|err| {
+        tracing::error!("Failed to sweep dead agents: {:?}", err);
+        format!("Database error: {}", err)
+    })?;
+
+    Ok(result.rows_affected())
+}
+
+/// Delete agents whose `deployment_hash` is structurally invalid.
+///
+/// A valid hash matches `deployment_<uuid>` (36-char UUID with hyphens).
+/// Rows with an invalid hash can never be matched by their own agent — the
+/// lookup in `fetch_by_deployment_hash` will never find them. Among the 18
+/// currently broken rows, 17 store a raw agent token (86-char base64url)
+/// instead of a hash, leaking the secret in plaintext.
+///
+/// Returns its own count separate from `sweep_dead` so the caller can log
+/// exactly how many malformed rows were removed.
+#[tracing::instrument(name = "Sweep malformed agent rows", skip(pool))]
+pub async fn sweep_malformed(pool: &PgPool) -> Result<u64, String> {
+    let result = sqlx::query(
+        r#"
+        DELETE FROM agents
+        WHERE deployment_hash !~ '^deployment_[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$'
+        "#,
+    )
+    .execute(pool)
+    .await
+    .map_err(|err| {
+        tracing::error!("Failed to sweep malformed agents: {:?}", err);
+        format!("Database error: {}", err)
+    })?;
+
+    Ok(result.rows_affected())
+}
+
 pub async fn log_audit(
     pool: &PgPool,
     audit_log: models::AuditLog,

src/services/agent_sweeper.rs

@@ -0,0 +1,76 @@
+//! Housekeeping for dead and malformed agent rows.
+//!
+//! The `agents` table accumulates rows that no longer serve a purpose: agents
+//! whose deployment was deleted or soft-deleted, agents that never sent a
+//! heartbeat, and rows whose `deployment_hash` is structurally invalid (e.g. a
+//! raw agent token stored instead of a hash). This sweeper removes them on a
+//! daily cadence.
+//!
+//! A row is considered "dead" only when **both** of the following hold:
+//!
+//!   1. The deployment is gone — either no matching row in `deployment`, or the
+//!      row exists with `deleted IS TRUE`.
+//!   2. There is no sign of life within the retention window — neither a
+//!      `last_heartbeat` nor an `audit_log` entry (the latter protects agents
+//!      that are alive but failing authentication, since `auth_failure` is
+//!      recorded in `audit_log` while `last_heartbeat` only advances on
+//!      successful `wait`/`report`).
+//!
+//! Malformed rows (invalid `deployment_hash`) are deleted unconditionally,
+//! without a retention window — such an agent can never authenticate by its own
+//! hash.
+//!
+//! Agents whose deployment is alive but silent are **not** touched: the row is
+//! the agent's identity, and removing it would require a full reinstall.
+
+use std::time::Duration;
+
+use sqlx::PgPool;
+
+use crate::db;
+
+/// How often to sweep. Rows appear rarely; daily is ample.
+const TICK: Duration = Duration::from_secs(86_400);
+
+/// How long a dead agent row is kept before removal. Long enough that a
+/// temporarily stopped server can come back without losing its identity.
+const RETENTION_DAYS: i32 = 30;
+
+pub fn spawn(pg_pool: PgPool) {
+    tokio::spawn(async move {
+        tracing::info!(
+            "agent_sweeper started (tick={:?}, retention={} days)",
+            TICK,
+            RETENTION_DAYS
+        );
+        loop {
+            // Sleep first: startup is busy enough, and nothing here is urgent.
+            tokio::time::sleep(TICK).await;
+
+            // Malformed rows are independent of retention — log at warn because
+            // new appearances indicate a write path that still needs fixing.
+            match db::agent::sweep_malformed(&pg_pool).await {
+                Ok(0) => {}
+                Ok(count) => tracing::warn!(
+                    "agent_sweeper: removed {} malformed agent row(s) — \
+                     the write path producing these has not been found yet",
+                    count
+                ),
+                Err(err) => {
+                    tracing::warn!("agent_sweeper: malformed sweep error: {}", err)
+                }
+            }
+
+            match db::agent::sweep_dead(&pg_pool, RETENTION_DAYS).await {
+                Ok(0) => tracing::debug!("agent_sweeper: nothing to remove"),
+                Ok(count) => tracing::info!(
+                    "agent_sweeper: removed {} dead agent row(s)",
+                    count
+                ),
+                Err(err) => {
+                    tracing::warn!("agent_sweeper: dead sweep error: {}", err)
+                }
+            }
+        }
+    });
+}

src/services/mod.rs

@@ -1,4 +1,5 @@
 pub mod agent_dispatcher;
+pub mod agent_sweeper;
 pub mod agent_token;
 pub mod config_renderer;
 pub mod dag_executor;

src/startup.rs

@@ -122,6 +122,11 @@ pub async fn run(
     // it touches nothing the dashboard shows, and skipping it costs a slowly
     // growing table rather than correctness.
     crate::services::deployment_container_sweeper::spawn(api_pool.get_ref().clone());
+    // Removes dead agent rows (no live deployment, no recent activity) and
+    // rows with structurally invalid deployment_hash. Runs daily; skipping it
+    // means the agents table slowly accumulates noise that obscures real
+    // problems.
+    crate::services::agent_sweeper::spawn(api_pool.get_ref().clone());
 
     let payout_provider = crate::services::init_payout_provider(&settings.payouts)
         .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidInput, err.to_string()))?;