gunnar_core/

activity.rs

//! **When a principal last signed in, and when it last changed something** —
//! the two answers an administrator reads in the console's user list
//! (owner, 2026-10-21).
//!
//! # Two stamps, kept apart
//!
//! * **signed in** — a console session was issued to the principal: a passkey
//!   login (`Access.WebLogin`, the SSO second factor that ends in one) or an
//!   SSO login that enrolled a session key.
//! * **operated** — an authenticated call that changed something: a push that
//!   moved a reference, or an allowed control-plane call that is not a read
//!   (a repository made or removed, a setting, a key added, …). A read is not
//!   an operation.
//!
//! # Where they live: a table of the control plane's redb, written in batches
//!
//! Not on the principal row: the identity table is ONE compare-and-swap
//! document (`refs/gunnar/auth`, see `AuthTable::touch_key`), and a write per
//! authenticated request would put every request on the box in contention for
//! one pointer. So the stamps are **kept in memory, moved at most once per
//! [`GRANULARITY`] per principal and kind**, and the rows that moved are
//! written by [`PrincipalActivity::flush`] — a background job
//! (a jera job of the server, `gunnar_server::stores::start_activity_flush`), never the request's — in **one redb
//! write transaction per flush that changed something**, into the control
//! database's own `principal_activity` table ([`crate::ControlDb`]). That
//! table is per box: never framed into the transaction log, never in a
//! snapshot or an export.
//!
//! **The twin holds them.** On a box that originates its transaction log, a
//! flush that moved something is also ONE [`crate::txlog::Kind::Activity`]
//! frame — the batch, appended before the table is written, under the log's
//! mutation guard — and the twin applies it into its own table
//! ([`PrincipalActivity::absorb`]). A snapshot carries the whole table as one
//! item, so a twin that joins later has every stamp; a twin promoted to main
//! keeps them. A follower frames nothing of its own.
//!
//! **Allowed to lose updates.** A crash loses at most the stamps since the last
//! flush ([`FLUSH_EVERY`]); a box restored from its twin starts with no stamps.
//! Either way the answer is "a little older than the truth" or "never", never
//! wrong about who. While the store is off the stripe (a data-set growth) a
//! flush is refused and the rows wait in memory for the next one.
//!
//! # The JSON file it replaced
//!
//! The first version wrote `principal-activity.json` beside the control
//! store ([`ACTIVITY_FILE`]). A box that has one reads it ONCE into the table
//! (the marker [`IMPORTED_MARK`] says it was) and leaves the file where it is.

use std::collections::{BTreeMap, BTreeSet};
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use serde::{Deserialize, Serialize};

use crate::control::ControlDb;
use crate::txlog::{Kind, Log, Meta};

/// The JSON file the stamps were kept in before they moved into redb; read
/// once into the table and never written again.
pub const ACTIVITY_FILE: &str = "principal-activity.json";

/// The marker (in the control database's per-box marks) that says
/// [`ACTIVITY_FILE`] has been read into the table.
pub const IMPORTED_MARK: &str = "principal-activity-json-imported";

/// How far a stamp must have moved before it is moved again: one minute, the
/// same grain as a token's `last_used` (`crate::auth::LAST_USED_GRANULARITY`).
/// The console reads it as "x minutes ago".
pub const GRANULARITY: i64 = 60_000;

/// How often the background thread writes what moved.
pub const FLUSH_EVERY: Duration = Duration::from_secs(30);

/// One principal's two stamps, Unix milliseconds. `None`: never (or not since
/// this box started keeping them).
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Seen {
    /// The last console session issued.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub signed_in: Option<i64>,
    /// The last authenticated call that changed something.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub operated: Option<i64>,
}

impl Seen {
    /// The later of each stamp.
    fn merge(&mut self, other: Seen) {
        self.signed_in = self.signed_in.max(other.signed_in);
        self.operated = self.operated.max(other.operated);
    }
}

#[derive(Default)]
struct Inner {
    seen: BTreeMap<String, Seen>,
    /// The principals whose stamps moved since the last flush.
    dirty: BTreeSet<String>,
}

/// The stamps, in memory, written to the control database in batches. See the
/// module.
pub struct PrincipalActivity {
    db: Option<Arc<ControlDb>>,
    /// The transaction log a flush frames its batch into (when it originates).
    log: Option<Arc<Log>>,
    inner: Mutex<Inner>,
}

/// **A batch of stamps as a frame's or a snapshot item's blob**: JSON,
/// principal id → stamps.
pub fn encode_batch(rows: &BTreeMap<String, Seen>) -> crate::error::Result<Vec<u8>> {
    serde_json::to_vec(rows).map_err(|err| crate::error::Error::ControlStore(format!("encoding last-seen stamps: {err}")))
}

/// [`encode_batch`] over the table's raw rows (each already one encoded [`Seen`]).
pub fn encode_rows(rows: &[(String, Vec<u8>)]) -> crate::error::Result<Vec<u8>> {
    let mut map = BTreeMap::new();
    for (id, raw) in rows {
        if let Ok(seen) = serde_json::from_slice::<Seen>(raw) {
            map.insert(id.clone(), seen);
        }
    }
    encode_batch(&map)
}

/// A frame's or a snapshot item's blob back to its stamps.
pub fn decode_batch(blob: &[u8]) -> crate::error::Result<BTreeMap<String, Seen>> {
    serde_json::from_slice(blob).map_err(|err| crate::error::Error::ControlStore(format!("a last-seen batch does not parse: {err}")))
}

fn row_of(id: &str, seen: &Seen) -> Option<(String, Vec<u8>)> {
    serde_json::to_vec(seen).ok().map(|v| (id.to_owned(), v))
}

impl std::fmt::Debug for PrincipalActivity {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PrincipalActivity").field("db", &self.db.is_some()).finish_non_exhaustive()
    }
}

/// Move `slot` to `now` when it is empty or at least [`GRANULARITY`] older.
/// Never backwards. `true` when it moved.
fn advance(slot: &mut Option<i64>, now: i64) -> bool {
    match *slot {
        Some(last) if now.saturating_sub(last) < GRANULARITY => false,
        _ => {
            *slot = Some(now);
            true
        }
    }
}

fn warn(what: &str, err: &dyn std::fmt::Display) {
    tracing::warn!(target: "gunnar::activity", error = %err, "{what}");
}

impl PrincipalActivity {
    /// Stamps kept in memory only (a test, a tool).
    pub fn in_memory() -> PrincipalActivity {
        PrincipalActivity { db: None, log: None, inner: Mutex::new(Inner::default()) }
    }

    /// The stamps kept in `db`'s `principal_activity` table, read back — and,
    /// the first time, the old [`ACTIVITY_FILE`] at `legacy_json` read into
    /// it (the file is left). A table that cannot be read, or a file that does
    /// not parse, is said at `warn!` and started over: this is bookkeeping,
    /// and a box must never refuse to start over it.
    pub fn open(db: Arc<ControlDb>, legacy_json: Option<PathBuf>, log: Option<Arc<Log>>) -> PrincipalActivity {
        let mut seen = BTreeMap::new();
        match db.principal_activity() {
            Ok(rows) => {
                for (id, raw) in rows {
                    match serde_json::from_slice::<Seen>(&raw) {
                        Ok(s) => {
                            seen.insert(id, s);
                        }
                        Err(err) => warn("a last-seen row does not parse; it is started over", &err),
                    }
                }
            }
            Err(err) => warn("the last-seen table could not be read; starting it over", &err),
        }
        let activity = PrincipalActivity { db: Some(Arc::clone(&db)), log, inner: Mutex::new(Inner { seen, dirty: BTreeSet::new() }) };
        if let Some(path) = legacy_json {
            activity.import_once(&db, path);
        }
        activity
    }

    /// Read the old JSON file into the table, once: its rows and the marker go
    /// in one transaction, and the file stays where it is.
    fn import_once(&self, db: &ControlDb, path: PathBuf) {
        match db.mark(IMPORTED_MARK) {
            Ok(Some(_)) => return,
            Ok(None) => {}
            Err(err) => return warn("the last-seen import marker could not be read", &err),
        }
        let old: BTreeMap<String, Seen> = match std::fs::read(&path) {
            Ok(bytes) => serde_json::from_slice(&bytes).unwrap_or_else(|err| {
                warn("the old principal-activity.json does not parse; it is not imported", &err);
                BTreeMap::new()
            }),
            // No old file: nothing to import, and nothing marked.
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => return,
            Err(err) => {
                warn("the old principal-activity.json could not be read; it is not imported", &err);
                BTreeMap::new()
            }
        };
        let rows: Vec<(String, Vec<u8>)> = {
            let mut inner = self.lock();
            for (id, s) in old {
                inner.seen.entry(id.clone()).or_default().merge(s);
                // The twin gets them with the next flush's frame.
                inner.dirty.insert(id);
            }
            let inner = &*inner;
            inner.seen.iter().filter_map(|(id, s)| row_of(id, s)).collect()
        };
        if let Err(err) = db.put_principal_activity(&rows, Some(IMPORTED_MARK)) {
            warn("the old principal-activity.json could not be written into the table; tried again next start", &err);
        }
    }

    fn lock(&self) -> std::sync::MutexGuard<'_, Inner> {
        self.inner.lock().unwrap_or_else(|p| p.into_inner())
    }

    fn stamp(&self, principal_id: &str, now: i64, pick: fn(&mut Seen) -> &mut Option<i64>) {
        if principal_id.is_empty() {
            return;
        }
        let mut inner = self.lock();
        // The common case — a stamp younger than the grain — allocates nothing.
        if let Some(seen) = inner.seen.get_mut(principal_id) {
            if advance(pick(seen), now) && !inner.dirty.contains(principal_id) {
                inner.dirty.insert(principal_id.to_owned());
            }
            return;
        }
        let mut seen = Seen::default();
        advance(pick(&mut seen), now);
        inner.seen.insert(principal_id.to_owned(), seen);
        inner.dirty.insert(principal_id.to_owned());
    }

    /// A console session was issued to `principal_id` at `now`.
    pub fn signed_in(&self, principal_id: &str, now: i64) {
        self.stamp(principal_id, now, |s| &mut s.signed_in);
    }

    /// `principal_id` made a call that changed something at `now`.
    pub fn operated(&self, principal_id: &str, now: i64) {
        self.stamp(principal_id, now, |s| &mut s.operated);
    }

    /// One principal's stamps.
    pub fn seen(&self, principal_id: &str) -> Seen {
        self.lock().seen.get(principal_id).copied().unwrap_or_default()
    }

    /// Every principal's stamps.
    pub fn all(&self) -> BTreeMap<String, Seen> {
        self.lock().seen.clone()
    }

    /// Something was stamped that is not in the table yet.
    pub fn is_dirty(&self) -> bool {
        !self.lock().dirty.is_empty()
    }

    /// Write the rows that moved since the last flush: **one** `Activity`
    /// frame for all of them (on a box that originates its log), then **one**
    /// redb write transaction. `Ok(false)` when nothing moved (nothing is
    /// written or framed) or this record keeps no database. A refused write
    /// keeps the rows for the next flush.
    pub fn flush(&self) -> crate::error::Result<bool> {
        let Some(db) = &self.db else {
            return Ok(false);
        };
        let (names, batch) = {
            let mut inner = self.lock();
            if inner.dirty.is_empty() {
                return Ok(false);
            }
            let names = std::mem::take(&mut inner.dirty);
            let batch: BTreeMap<String, Seen> =
                names.iter().filter_map(|id| inner.seen.get(id).map(|s| (id.clone(), *s))).collect();
            (names, batch)
        };
        let rows: Vec<(String, Vec<u8>)> = batch.iter().filter_map(|(id, s)| row_of(id, s)).collect();
        let written = (|| {
            let framing = self.log.as_ref().filter(|log| !log.is_follower());
            // The frame and the table under one mutation guard: a snapshot's cut sees both or
            // neither.
            let _m = framing.map(|log| log.mutation());
            if let Some(log) = framing {
                log.append(Kind::Activity, &Meta::Activity {}, &encode_batch(&batch)?)
                    .map_err(|err| crate::error::Error::ControlStore(format!("framing last-seen stamps: {err}")))?;
            }
            db.put_principal_activity(&rows, None)
        })();
        if let Err(err) = written {
            self.lock().dirty.extend(names);
            return Err(err);
        }
        Ok(true)
    }

    /// **Take stamps from the primary** (a twin applying an `Activity` frame or a
    /// snapshot item; a box being restored): each one merged in memory, the later
    /// of each stamp kept, and the merged rows written to the table in one
    /// transaction. Framed by nobody — it is already a frame.
    pub fn absorb(&self, rows: BTreeMap<String, Seen>) -> crate::error::Result<()> {
        let merged: Vec<(String, Vec<u8>)> = {
            let mut inner = self.lock();
            rows.into_iter()
                .filter_map(|(id, s)| {
                    let slot = inner.seen.entry(id.clone()).or_default();
                    slot.merge(s);
                    row_of(&id, slot)
                })
                .collect()
        };
        match &self.db {
            Some(db) if !merged.is_empty() => db.put_principal_activity(&merged, None),
            _ => Ok(()),
        }
    }

    fn flush_logged(&self) {
        if let Err(err) = self.flush() {
            warn("the last-seen rows could not be written; they are kept in memory and tried again", &err);
        }
    }
}

impl Drop for PrincipalActivity {
    fn drop(&mut self) {
        self.flush_logged();
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn db_at(path: &std::path::Path) -> Arc<ControlDb> {
        crate::control::open_control_with_db(path).expect("control store").2
    }

    #[test]
    fn a_stamp_moves_only_past_the_grain_and_never_backwards() {
        let a = PrincipalActivity::in_memory();
        a.operated("kim", 1_000_000);
        assert_eq!(a.seen("kim").operated, Some(1_000_000));
        a.operated("kim", 1_000_000 + GRANULARITY - 1);
        assert_eq!(a.seen("kim").operated, Some(1_000_000), "inside the grain");
        a.operated("kim", 1_000_000 - 5 * GRANULARITY);
        assert_eq!(a.seen("kim").operated, Some(1_000_000), "an older stamp does not move it back");
        a.operated("kim", 1_000_000 + GRANULARITY);
        assert_eq!(a.seen("kim").operated, Some(1_000_000 + GRANULARITY));
        assert_eq!(a.seen("kim").signed_in, None, "the two stamps are apart");
        assert_eq!(a.seen("nobody"), Seen::default());
    }

    #[test]
    fn a_flush_is_one_commit_for_many_stamps_and_none_when_nothing_moved() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("control.redb");
        {
            let db = db_at(&path);
            let a = PrincipalActivity::open(Arc::clone(&db), None, None);
            let before = db.durable_commits();
            assert!(!a.flush().expect("flush"), "nothing moved, nothing written");
            a.signed_in("kim", 5_000_000);
            for i in 0..1_000 {
                a.operated("kim", 5_000_000 + i);
            }
            a.operated("ann", 6_000_000);
            assert_eq!(db.durable_commits(), before, "a stamp writes nothing");
            assert!(a.flush().expect("flush"));
            assert_eq!(db.durable_commits(), before + 1, "one transaction for the whole batch");
            assert!(!a.flush().expect("flush"), "the second flush has nothing to write");
            a.operated("kim", 5_000_000 + 10);
            assert!(!a.is_dirty(), "a stamp inside the grain does not make a write");
            assert_eq!(db.durable_commits(), before + 1);
        }
        let a = PrincipalActivity::open(db_at(&path), None, None);
        assert_eq!(a.seen("kim"), Seen { signed_in: Some(5_000_000), operated: Some(5_000_000) });
        assert_eq!(a.seen("ann").operated, Some(6_000_000));
    }

    #[test]
    fn the_last_stamps_are_written_when_the_record_is_dropped() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("control.redb");
        PrincipalActivity::open(db_at(&path), None, None).signed_in("kim", 7);
        assert_eq!(PrincipalActivity::open(db_at(&path), None, None).seen("kim").signed_in, Some(7));
    }

    #[test]
    fn an_old_json_file_is_imported_once_and_left() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("control.redb");
        let json = dir.path().join(ACTIVITY_FILE);
        std::fs::write(&json, br#"{"kim":{"signed_in":1000,"operated":2000}}"#).expect("write");
        assert_eq!(PrincipalActivity::open(db_at(&path), Some(json.clone()), None).seen("kim"), Seen { signed_in: Some(1000), operated: Some(2000) });
        assert!(json.exists(), "the file is left");
        std::fs::write(&json, br#"{"kim":{"signed_in":9000}}"#).expect("rewrite");
        assert_eq!(PrincipalActivity::open(db_at(&path), Some(json), None).seen("kim").signed_in, Some(1000), "read once");
    }

    #[test]
    fn a_flush_on_an_originating_box_is_one_frame_and_a_follower_frames_nothing() {
        let dir = tempfile::tempdir().expect("tempdir");
        let log = Arc::new(Log::open(dir.path().join("txlog")).expect("log"));
        let a = PrincipalActivity::open(db_at(&dir.path().join("control.redb")), None, Some(Arc::clone(&log)));
        a.signed_in("kim", 1);
        a.operated("ann", 2);
        assert!(a.flush().expect("flush"));
        assert_eq!(log.head_seq(), 1, "one frame for the batch");
        let e = log.read(1).expect("the frame");
        assert_eq!(e.kind(), Some(Kind::Activity));
        let batch = decode_batch(&e.blob).expect("decodes");
        assert_eq!(batch.get("kim").and_then(|s| s.signed_in), Some(1));
        assert_eq!(batch.get("ann").and_then(|s| s.operated), Some(2));
        log.set_follower(true);
        a.operated("kim", 10 * GRANULARITY);
        assert!(a.flush().expect("a follower still writes its table"));
        assert_eq!(log.head_seq(), 1, "a follower frames nothing");
    }

    #[test]
    fn absorbed_stamps_are_merged_and_written_without_a_frame() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("control.redb");
        {
            let a = PrincipalActivity::open(db_at(&path), None, None);
            a.operated("kim", 5_000_000);
            a.absorb(BTreeMap::from([("kim".to_owned(), Seen { signed_in: Some(7), operated: Some(1) })])).expect("absorb");
            assert_eq!(a.seen("kim"), Seen { signed_in: Some(7), operated: Some(5_000_000) }, "the later of each");
        }
        assert_eq!(PrincipalActivity::open(db_at(&path), None, None).seen("kim").signed_in, Some(7));
    }

    #[test]
    fn a_damaged_old_file_is_not_fatal() {
        let dir = tempfile::tempdir().expect("tempdir");
        let json = dir.path().join(ACTIVITY_FILE);
        std::fs::write(&json, b"{not json").expect("write");
        assert!(PrincipalActivity::open(db_at(&dir.path().join("control.redb")), Some(json), None).all().is_empty());
    }
}