feat(security): enforce per-database idle timeout and typed kill reasons

Introduces KillReason, an idle-session sweep loop, per-database idle
timeout configuration, and richer SHOW SESSIONS output.

KillReason enum replaces the bool kill signal on the session watch
channel, allowing the session drop path to surface a precise error
(IdleTimeout, TokenExpired, AdminKill, UserDropped) rather than a
generic "killed" state.

IdleTimeoutCache caches the idle_session_timeout_secs per database in
memory so the sweep loop avoids catalog round-trips. Populated at
startup from the catalog and updated synchronously by the ALTER DATABASE
SET IDLE_TIMEOUT handler.

ALTER DATABASE SET IDLE_TIMEOUT = <secs> (0 = disable) is gated on
ClusterAdmin/Superuser, writes the updated descriptor to the catalog,
updates the live IdleTimeoutCache, and emits a
DatabaseIdleTimeoutChanged audit event.

idle_sweep::sweep_once() runs on a 5-second Tokio tick, snapshots
sessions via sweep_snapshot(), computes the earlier of the per-database
idle deadline and the OIDC token expiry deadline, and signals the
appropriate KillReason. The sweep loop never calls unregister(); only
the session's own drop path does.

SessionRegistry gains:
- current_database, idle_timeout_secs, token_expiry_ms, bytes_in,
  bytes_out, current_statement_digest fields on RegisteredSession.
- kill_session_by_id() for targeted one-shot kill with a KillReason.
- lookup_session_database() for pre-kill authority checks.
- sweep_snapshot() producing SweepEntry rows consumed by the sweep loop.

SHOW SESSIONS gains idle_seconds, current_database, bytes_in,
bytes_out, current_statement, and token_expires_in_secs columns.

KILL SESSION is extended to allow ClusterAdmin and DatabaseOwner of the
session's database to kill sessions (not just Superuser). A race
between authority check and kill is surfaced as a distinct audit row.

kill_sessions_for_user() and kill_sessions_for_username() now accept a
KillReason parameter so callers (bus consumer, blacklist DDL, emergency
DDL) can declare intent rather than relying on the sentinel bool.

Author: farhan-syah

nodedb/src/bootstrap/background_loops.rs

@@ -198,6 +198,11 @@ pub fn spawn_background_loops(
         },
     );
 
+    // Idle session sweep (5-second timer).
+    // Closes sessions whose idle timeout or OIDC token expiry has elapsed.
+    crate::control::security::sessions::spawn_idle_sweep_loop(shared);
+    info!("idle session sweep loop running");
+
     // Audit log flush (10-second timer).
     let shared_audit = Arc::clone(shared);
     crate::control::shutdown::spawn_loop(

nodedb/src/control/mirror/restart.rs

@@ -132,6 +132,7 @@ mod tests {
                 status,
             }),
             audit_dml: nodedb_types::AuditDmlMode::None,
+            idle_session_timeout_secs: 0,
         }
     }
 

nodedb/src/control/security/audit/event.rs

@@ -133,6 +133,11 @@ pub enum AuditEvent {
     /// The AUDIT_DML mode for a database was changed via
     /// `ALTER DATABASE SET AUDIT_DML`.
     DatabaseAuditDmlChanged = 44,
+    /// The idle session timeout for a database was changed via
+    /// `ALTER DATABASE SET IDLE_TIMEOUT`.
+    DatabaseIdleTimeoutChanged = 45,
+    /// An OIDC provider was created, altered, or dropped.
+    OidcProviderChanged = 46,
 }
 
 impl AuditEvent {
@@ -187,6 +192,8 @@ impl AuditEvent {
             Self::DatabaseRestored => 42,
             Self::DmlAudit => 43,
             Self::DatabaseAuditDmlChanged => 44,
+            Self::DatabaseIdleTimeoutChanged => 45,
+            Self::OidcProviderChanged => 46,
         }
     }
 
@@ -237,7 +244,9 @@ impl AuditEvent {
             | Self::DatabaseBackedUp
             | Self::DatabaseRestored
             | Self::DmlAudit
-            | Self::DatabaseAuditDmlChanged => false,
+            | Self::DatabaseAuditDmlChanged
+            | Self::DatabaseIdleTimeoutChanged
+            | Self::OidcProviderChanged => false,
         }
     }
 
@@ -291,6 +300,8 @@ impl AuditEvent {
             Self::DatabaseRestored => "database_restored",
             Self::DmlAudit => "dml_audit",
             Self::DatabaseAuditDmlChanged => "database_audit_dml_changed",
+            Self::DatabaseIdleTimeoutChanged => "database_idle_timeout_changed",
+            Self::OidcProviderChanged => "oidc_provider_changed",
         }
     }
 
@@ -341,7 +352,9 @@ impl AuditEvent {
             | Self::TenantMoved
             | Self::DatabaseBackedUp
             | Self::DatabaseRestored
-            | Self::DatabaseAuditDmlChanged => AuditLevel::Standard,
+            | Self::DatabaseAuditDmlChanged
+            | Self::DatabaseIdleTimeoutChanged
+            | Self::OidcProviderChanged => AuditLevel::Standard,
             Self::DmlAudit => AuditLevel::Forensic,
         }
     }

nodedb/src/control/security/buses/consumer.rs

@@ -157,7 +157,10 @@ fn handle_session_invalidated(
 
     // Hard revoke: signal every open session for this user to close.
     if event.reason.is_hard_revoke() {
-        session_registry.kill_sessions_for_user(event.user_id);
+        session_registry.kill_sessions_for_user(
+            event.user_id,
+            crate::control::security::sessions::KillReason::UserDropped,
+        );
     }
     // Soft revoke: identity rehydrate is driven by the per-user version
     // counter at the next request-entry boundary — no explicit signal needed.
@@ -240,6 +243,8 @@ mod tests {
             auth_method: "password".into(),
             tenant_id: 1,
             credential_version: 0,
+            current_database: None,
+            token_expiry_ms: None,
         };
         let mut kill_rx = registry.register("session-42", &params).unwrap();
 
@@ -266,7 +271,11 @@ mod tests {
             kill_rx.has_changed().unwrap_or(false),
             "kill_rx must be signalled for hard revoke"
         );
-        assert!(*kill_rx.borrow_and_update(), "kill value must be true");
+        assert_ne!(
+            *kill_rx.borrow_and_update(),
+            crate::control::security::sessions::KillReason::Alive,
+            "kill value must not be Alive"
+        );
     }
 
     #[tokio::test]

nodedb/src/control/security/sessions/idle_sweep.rs

@@ -0,0 +1,202 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Idle-session sweep loop (Control Plane, Tokio).
+//!
+//! `idle_sweep_loop` runs on the Tokio runtime at a 5-second tick.  Each tick
+//! it calls `sweep_once`, which:
+//!
+//! 1. Takes a lock-free snapshot of every session via `sweep_snapshot`.
+//! 2. For each session, computes the earliest close deadline from:
+//!    - The per-database idle timeout (from the session's cached
+//!      `idle_timeout_secs`, which mirrors the `IdleTimeoutCache`).
+//!    - The OIDC token expiry (`token_expiry_ms`), if set.
+//!    - The global idle timeout stored in `SharedState::idle_timeout_secs`.
+//! 3. If `now_ms >= deadline`, sends `KillReason::IdleTimeout` or
+//!    `KillReason::TokenExpired` (whichever is earlier) to that session.
+//!
+//! The session's own drop path calls `unregister` — the sweep loop only signals
+//! via the `kill_tx`; it never calls `unregister` itself.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use crate::control::security::sessions::{KillReason, SessionRegistry};
+use crate::control::security::time::now_ms;
+use crate::control::state::SharedState;
+
+/// Compute the next close deadline (milliseconds since epoch) for a session,
+/// given:
+/// - `token_exp_ms`: OIDC token expiry in ms (0 = no token expiry).
+/// - `last_active_ms`: last-active timestamp in milliseconds.
+/// - `idle_cap_secs`: idle timeout in seconds (0 = no idle cap).
+///
+/// Returns `None` if neither deadline applies.
+/// Returns the earlier of the two deadlines when both apply.
+pub fn next_close_deadline_ms(
+    token_exp_ms: u64,
+    last_active_ms: u64,
+    idle_cap_secs: u64,
+) -> Option<u64> {
+    let idle_deadline = if idle_cap_secs > 0 {
+        Some(last_active_ms.saturating_add(idle_cap_secs * 1_000))
+    } else {
+        None
+    };
+    let token_deadline = if token_exp_ms > 0 {
+        Some(token_exp_ms)
+    } else {
+        None
+    };
+    match (idle_deadline, token_deadline) {
+        (Some(a), Some(b)) => Some(a.min(b)),
+        (Some(a), None) => Some(a),
+        (None, Some(b)) => Some(b),
+        (None, None) => None,
+    }
+}
+
+/// Determine the kill reason for a deadline breach, preferring
+/// `TokenExpired` when the token deadline is earlier than the idle deadline.
+fn kill_reason_for(
+    token_exp_ms: u64,
+    last_active_ms: u64,
+    idle_cap_secs: u64,
+    now_ms: u64,
+) -> KillReason {
+    // Evaluate which deadline fired first.
+    let token_fired = token_exp_ms > 0 && now_ms >= token_exp_ms;
+    let idle_fired =
+        idle_cap_secs > 0 && now_ms >= last_active_ms.saturating_add(idle_cap_secs * 1_000);
+
+    match (token_fired, idle_fired) {
+        (true, false) => KillReason::TokenExpired,
+        (false, true) => KillReason::IdleTimeout,
+        (true, true) => {
+            // Both fired: prefer the one with the earlier deadline.
+            let token_dl = token_exp_ms;
+            let idle_dl = last_active_ms.saturating_add(idle_cap_secs * 1_000);
+            if token_dl <= idle_dl {
+                KillReason::TokenExpired
+            } else {
+                KillReason::IdleTimeout
+            }
+        }
+        (false, false) => KillReason::Alive,
+    }
+}
+
+/// Single sweep pass.
+///
+/// Exported for testing; production code calls this via `idle_sweep_loop`.
+pub fn sweep_once(session_registry: &Arc<SessionRegistry>, global_idle_secs: u64) {
+    let now = now_ms();
+    let entries = session_registry.sweep_snapshot();
+    for entry in entries {
+        // Per-session cap: the session's cached `idle_timeout_secs` (loaded from
+        // `IdleTimeoutCache` at session start and on cache invalidation).
+        // Falls back to the global cap when the per-database value is 0.
+        let idle_secs = if entry.idle_timeout_secs > 0 {
+            entry.idle_timeout_secs
+        } else {
+            global_idle_secs
+        };
+        // last_active is stored in seconds; convert to ms for deadline arithmetic.
+        let last_active_ms = entry.last_active_secs * 1_000;
+        let deadline = next_close_deadline_ms(entry.token_expiry_ms, last_active_ms, idle_secs);
+        let Some(dl) = deadline else {
+            continue;
+        };
+        if now >= dl {
+            let reason = kill_reason_for(entry.token_expiry_ms, last_active_ms, idle_secs, now);
+            if reason != KillReason::Alive {
+                let _ = session_registry.kill_session_by_id(&entry.session_id, reason);
+            }
+        }
+    }
+}
+
+/// Spawn the idle sweep loop on the Tokio runtime.
+///
+/// Runs at a 5-second tick.  Respects `SharedState::shutdown` for graceful
+/// termination.  Registered in `loop_registry` for drain-on-shutdown.
+pub fn spawn_idle_sweep_loop(shared: &Arc<SharedState>) {
+    let shared_sweep = Arc::clone(shared);
+    crate::control::shutdown::spawn_loop(
+        &shared.loop_registry,
+        &shared.shutdown,
+        "idle_session_sweep",
+        move |mut shutdown| async move {
+            let mut tick = tokio::time::interval(Duration::from_secs(5));
+            loop {
+                tokio::select! {
+                    _ = shutdown.wait_cancelled() => break,
+                    _ = tick.tick() => {}
+                }
+                if shutdown.is_cancelled() {
+                    break;
+                }
+                let global_idle = shared_sweep.idle_timeout_secs();
+                sweep_once(&shared_sweep.session_registry, global_idle);
+            }
+        },
+    );
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn no_deadline_when_both_zero() {
+        assert_eq!(next_close_deadline_ms(0, 1_000_000, 0), None);
+    }
+
+    #[test]
+    fn idle_deadline_only() {
+        // last_active_ms = 1000s, cap = 300s → deadline = 1300s in ms
+        let dl = next_close_deadline_ms(0, 1_000_000, 300);
+        assert_eq!(dl, Some(1_300_000));
+    }
+
+    #[test]
+    fn token_deadline_only() {
+        let dl = next_close_deadline_ms(9_999_000, 1_000_000, 0);
+        assert_eq!(dl, Some(9_999_000));
+    }
+
+    #[test]
+    fn picks_earlier_idle_over_token() {
+        // idle fires at 1_300_000, token at 9_999_000 → idle wins
+        let dl = next_close_deadline_ms(9_999_000, 1_000_000, 300);
+        assert_eq!(dl, Some(1_300_000));
+    }
+
+    #[test]
+    fn picks_earlier_token_over_idle() {
+        // token fires at 500_000, idle at 1_300_000 → token wins
+        let dl = next_close_deadline_ms(500_000, 1_000_000, 300);
+        assert_eq!(dl, Some(500_000));
+    }
+
+    #[test]
+    fn kill_reason_token_when_token_earlier() {
+        let now = 600_000u64;
+        let reason = kill_reason_for(500_000, 1_000_000, 300, now);
+        assert_eq!(reason, KillReason::TokenExpired);
+    }
+
+    #[test]
+    fn kill_reason_idle_when_idle_fires() {
+        let now = 1_400_000u64;
+        // idle deadline = 1_300_000, token at 9_999_000 → idle
+        let reason = kill_reason_for(9_999_000, 1_000_000, 300, now);
+        assert_eq!(reason, KillReason::IdleTimeout);
+    }
+
+    #[test]
+    fn alive_when_not_expired() {
+        let now = 100_000u64;
+        let reason = kill_reason_for(9_999_000, 1_000_000, 300, now);
+        assert_eq!(reason, KillReason::Alive);
+    }
+}

nodedb/src/control/security/sessions/mod.rs

@@ -1,7 +1,31 @@
 // SPDX-License-Identifier: BUSL-1.1
 
-//! Active session tracking (registry, cap enforcement, revocation).
+//! Active session tracking: registry, cap enforcement, revocation, idle sweep.
+//!
+//! ## Lifecycle
+//!
+//! `SessionRegistry` is constructed once during server startup and shared
+//! via `SharedState::session_registry`. Every authenticated connection
+//! (native, pgwire, HTTP) calls `register` on bind and `unregister` on drop.
+//!
+//! `spawn_idle_sweep_loop` MUST be called exactly once at server startup,
+//! after `SharedState` is fully built. It registers a Tokio task with the
+//! shared `loop_registry` (so it drains on shutdown) and ticks every five
+//! seconds, signalling `KillReason::IdleTimeout` or `KillReason::TokenExpired`
+//! to sessions whose per-database idle cap or OIDC token expiry has elapsed.
+//! Calling it more than once would spawn duplicate sweepers — there is no
+//! internal idempotency guard, so the call site (currently
+//! `nodedb::bootstrap::background_loops`) is responsible for the
+//! exactly-once invariant.
+//!
+//! Callers never invoke `unregister` from the sweep path; the sweep loop
+//! only sends on `kill_tx`, and the session's own drop path removes the
+//! row.
 
+pub mod idle_sweep;
 pub mod registry;
 
-pub use registry::{SessionCapExceeded, SessionInfo, SessionParams, SessionRegistry};
+pub use idle_sweep::spawn_idle_sweep_loop;
+pub use registry::{
+    KillReason, SessionCapExceeded, SessionInfo, SessionParams, SessionRegistry, SweepEntry,
+};