crates/gunnar-core/src/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()
);
}
}