diff --git a/Cargo.toml b/Cargo.toml index 346ba04a..6f3d85ba 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,7 +10,7 @@ mimalloc = { version = "0.1.52", default-features = false } axum = { version = "0.8.9", features = ["multipart", "http1", "http2", "tokio", "macros"] } # "process" was previously enabled implicitly through aws-config's feature # unification; ffmpeg_video_frame_service needs it, so declare it ourselves. -tokio = { version = "1.52.3", features = ["rt-multi-thread", "macros", "io-util", "net", "time", "sync", "fs", "process"] } +tokio = { version = "1.52.3", features = ["rt-multi-thread", "macros", "io-util", "net", "time", "sync", "fs", "process", "signal"] } tokio-util = { version = "0.7.18", features = ["io", "codec", "compat"] } tokio-stream = { version = "0.1.18", features = ["fs", "sync"] } bytes = "1.11.1" diff --git a/docs/plan/sessions.md b/docs/plan/sessions.md new file mode 100644 index 00000000..1ce0c0cc --- /dev/null +++ b/docs/plan/sessions.md @@ -0,0 +1,401 @@ +# Session Liveness Tracking — `last_seen_at` + Prometheus + +Track per-session and per-user "currently active" signals cheaply, and +expose them as Prometheus gauges so an operator (demo instance, +production) can plot concurrency over time. Also unlocks future +features that need "when was this session last used" (idle-timeout +enforcement, per-user session-limit quotas, admin dashboard freshness). + +Companion doc for the design decisions covered here — narrative on the +overall session model lives in +[docs/architecture/auth-model.md](../architecture/auth-model.md). + +## Purpose — what we want to see + +Two distinct signals, deliberately separate: + +1. **Online sessions** (`oxicloud_sessions_online`) — count of + non-revoked `auth.sessions` rows that had a request within the last + N minutes. One user with three devices (browser + phone + Nextcloud + desktop) contributes **three** to this count. Useful for + provisioning ("how many concurrent connections do I need to + support?") and load-shape planning. + +2. **Online users** (`oxicloud_sessions_online_users`) — count of + DISTINCT `user_id` values behind those online sessions. Same + three-device user contributes **one** to this count. Useful for + billing shape ("how many humans are actually using the system?") + and for the demo landing page's "N users online right now" widget. + +The gap between the two IS the multi-device factor. A healthy system +where users routinely have web + desktop client should show `sessions +≈ 2 × users`. A sudden `sessions >> users × 3` is a signal — an app +that opens fresh sessions instead of reusing them, or a +credential-stuffing pattern. + +### Terminology — "online" vs "active" + +The admin sessions panel already has a lifecycle filter +`Active | Expired | Revoked` — where **active** means +`!revoked && !expired` (row is still usable). That's orthogonal to +"had a request lately", so both concepts fighting for the same word +was going to confuse admins reading the panel. + +**Decision (Ed, 2026-08-18)** — "online" is the *presence* signal +throughout the stack: + +- **UI**: green-dot badge next to each row when + `SessionSummaryDto::is_online == true`; grey dot + "last seen X ago" + otherwise. Lifecycle filter stays `Active | Expired | Revoked` + unchanged. +- **DTO**: `is_online: bool` on `SessionSummaryDto`, computed + server-side (avoids the SPA doing clock math and drifting from the + server view). Guaranteed `false` on revoked / expired rows so an + admin never sees "Online" on a row they just revoked. +- **Metrics**: `oxicloud_sessions_online` / `_online_users`. +- **Threshold**: single `pub const ONLINE_WINDOW` in + `application/dtos/session_dto.rs` — DTO derivation AND gauge query + read from the same constant so the per-row badge count and the + gauge aggregate stay consistent by construction. + +## Why the existing signals don't answer this — DECIDED + +`auth.sessions.created_at` is the closest existing proxy. But it +moves on **session rotation**, not per-request: + +- Sessions rotate on every silent refresh (`apiFetch`'s 401 → refresh + path). Rotation cadence = `access_token_expiry_secs` (default + 3600, i.e. 1 h). +- So `WHERE created_at > NOW() - INTERVAL '1 hour'` catches everyone + who refreshed in the last cycle — but a user who's actively clicking + around for 45 min hasn't rotated yet, so their `created_at` is 45 + min old. Threshold `< 30 min` false-negatives them. +- Resolution is capped at the access-token TTL. At the recommended + prod value of 15 min, `created_at` gives 15-min granularity. At the + test value of 60 s it's near-real-time — but no operator wants + to force 60 s token TTL just for observability. + +So `created_at` is an OK first-pass proxy but bad enough that we +should add a dedicated column that moves per-request. + +## Schema — `last_seen_at` + +Migration `_sessions_last_seen_at.sql`: + +```sql +ALTER TABLE auth.sessions + ADD COLUMN last_seen_at TIMESTAMPTZ NOT NULL DEFAULT NOW(); + +-- Partial index — the "active in the last N min" query is the only +-- reason to scan this column, and it always filters on revoked = FALSE. +-- Full index would double the write cost for zero read benefit. +CREATE INDEX idx_sessions_last_seen_at ON auth.sessions(last_seen_at) + WHERE revoked = FALSE; +``` + +Default `NOW()` — existing rows on migration land at "just seen" +which is slightly optimistic but the alternative (NULL / epoch) makes +every historic session count as long-idle for the first N min after +deploy. `NOW()` matches "assume everyone's active" and the metric +converges to reality within one N-min bucket. + +**Not in scope for this schema**: adding `last_seen_at` to +`auth.app_passwords`. Nextcloud-desktop clients authenticate through +that table; separate concern, separate PR if we want desktop-client +liveness. + +## Coalescing writes — DECIDED: in-process DashMap + periodic flush + +Naive shape (one `UPDATE` per authenticated request) would multiply +the DB write rate by every non-mutating request the SPA fires +(listing pages, thumbnails, delta-upload chunk PUTs). Untenable. + +The pattern that scales: + +- Middleware stamps `(session_id, Instant::now())` into a shared + `DashMap>`. **O(1)** per request, no I/O. +- Background task drains the map every N seconds (default 30) and + emits ONE batched `UPDATE ... FROM UNNEST(...)` covering every + distinct session seen in the window. +- The map data structure IS the dedup: same `session_id` → same key + → last write wins. A session seen 100× in the window contributes + ONE row to the batched update, with the latest timestamp. + +The `UPDATE` uses `greatest()` to be idempotent under any +retry/race/clock-skew: + +```sql +UPDATE auth.sessions AS s +SET last_seen_at = greatest(s.last_seen_at, t.seen_at) +FROM UNNEST($1::uuid[], $2::timestamptz[]) AS t(id, seen_at) +WHERE s.id = t.id; +``` + +**Restart durability**: up to one flush interval of activity lost on +hard crash. On graceful shutdown (SIGTERM) run one final flush +synchronously before exit — zero loss on planned rolling restarts. + +**Failure durability**: if the batched UPDATE fails (PG blip), +DON'T clear the DashMap; next tick retries with the accumulated set +overlaid on any new activity. + +### Why not `Arc>` + +Every authenticated request writes. Under any concurrency (delta +uploads, thumbnail bursts, folder listing paginations firing in +parallel) a single mutex becomes the bottleneck. `DashMap`'s +per-shard locking (16-32 shards by default) parallelises writes +across distinct keys — different session_ids don't contend. + +### Why not `Arc>` + +The workload is write-heavy (every auth'd request writes, reads only +fire in the flusher). RwLock would still serialize the writes for no +benefit. + +### Why not PG `NOTIFY` / `LISTEN` + +Considered. Trade-offs: + +- **Pro**: cross-instance coalescing — multiple OxiCloud processes + behind a load balancer push to one channel, single flusher owns + writes. +- **Con**: every request pays a PG round-trip (`SELECT + pg_notify(...)`) — ~1 ms per request vs ~50 ns for DashMap insert. + On hot endpoints (delta chunk PUTs, thumbnails) this is + measurable. +- **Con**: adds a persistent LISTEN connection to the pool. +- **Con**: OxiCloud is single-instance today. The multi-instance win + doesn't apply. + +**Deferred**: if OxiCloud ever grows a multi-instance deployment +story (Kubernetes, active-active behind a load balancer), migrate +the flusher to `NOTIFY`-based ingest — schema stays identical, only +the tracker implementation swaps. Document the migration path in +[Future — multi-instance](#future--multi-instance) below. + +## Metric surface — Prometheus + +Exposed via the existing `/metrics` endpoint (see +`src/interfaces/metrics.rs`; gated on `OXICLOUD_METRICS_LISTEN`). + +### Gauges (polled every 30 s from a background task) + +``` +# HELP oxicloud_sessions_online Non-revoked sessions seen in the last N min. +# TYPE oxicloud_sessions_online gauge +oxicloud_sessions_online + +# HELP oxicloud_sessions_online_users Distinct users behind online sessions. +# TYPE oxicloud_sessions_online_users gauge +oxicloud_sessions_online_users + +# HELP oxicloud_sessions_total_non_revoked Total non-revoked sessions +# regardless of activity — the long tail (mobile clients still holding +# refresh tokens they haven't used in weeks). +# TYPE oxicloud_sessions_total_non_revoked gauge +oxicloud_sessions_total_non_revoked +``` + +Queries powering each: + +```sql +-- oxicloud_sessions_online +SELECT COUNT(*) FROM auth.sessions +WHERE revoked = FALSE + AND last_seen_at > NOW() - $1::interval; -- $1 = ONLINE_WINDOW + +-- oxicloud_sessions_online_users +SELECT COUNT(DISTINCT user_id) FROM auth.sessions +WHERE revoked = FALSE + AND last_seen_at > NOW() - $1::interval; + +-- oxicloud_sessions_total_non_revoked +SELECT COUNT(*) FROM auth.sessions WHERE revoked = FALSE; +``` + +All three run on the maintenance pool (background polling shouldn't +compete with request-serving connections). Three lightweight +`COUNT(*)` reads every 30 s; measured cost negligible even on +tens-of-thousands-of-rows tables thanks to the partial index. + +### Counters (already in-place shape) + +`oxicloud_sessions_created_total` and +`oxicloud_sessions_revoked_total{reason}` — extend the existing +counter surface in the auth service (`session.created` audit line +sites) to also `metrics::counter!(...)`. Not strictly needed for the +"how many active" question but useful sanity signal on the +dashboard: rate of creation vs rate of revocation should be +approximately balanced at steady state. + +## Config surface + +**No new env var.** Ed's call (2026-08-18): tuning the online window +is a deployment-shape question we haven't had to answer in practice, +and adding an env knob invites premature customization. The three +knobs stay hardcoded: + +- **Online window** — 5 min. Feels responsive for a demo + landing page without over-fluctuating with tab-open-then-close + blips. Lives at `pub const ONLINE_WINDOW` in + `src/application/dtos/session_dto.rs`; the gauges module in + `src/infrastructure/services/session_liveness_gauges.rs` reads + from that constant so the DTO badge and the gauge aggregate + can't drift. +- **Flush interval** — 30 s. Balances DB write load against gauge + freshness (typical Prometheus scrape at 15 s sees the value + refreshed after at most two scrapes). Lives at `FLUSH_INTERVAL` + in `src/infrastructure/services/last_seen_tracker.rs`. +- **DashMap shard count** — crate default (16). Only worth + surfacing when profiling shows shard contention. + +## Middleware wiring + +Auth extractor (`CurrentUserId`) already loads the session by +refresh-token cookie / bearer-token subject. Extend the post-load +path: + +```rust +// After successful session lookup + auth checks: +state.last_seen_tracker.stamp(session.id); +``` + +`LastSeenTracker` shape: + +```rust +pub struct LastSeenTracker { + seen: Arc>>, + pool: Arc, +} + +impl LastSeenTracker { + pub fn new(pool: Arc) -> Arc { + let seen = Arc::new(DashMap::new()); + let this = Arc::new(Self { seen: seen.clone(), pool: pool.clone() }); + tokio::spawn(this.clone().flush_loop()); + this + } + + /// Called from the auth middleware on every authenticated request. + /// O(1); no I/O; no round-trip. + pub fn stamp(&self, session_id: Uuid) { + self.seen.insert(session_id, Utc::now()); + } + + /// Called from the graceful-shutdown handler. + pub async fn flush_now(&self) -> Result<(), sqlx::Error> { + /* drain + one batched UPDATE, same as the loop body */ + } + + async fn flush_loop(self: Arc) { + let mut ticker = tokio::time::interval(Duration::from_secs(30)); + loop { + ticker.tick().await; + let _ = self.flush_now().await; // errors logged, not propagated + } + } +} +``` + +Wired in `common/di.rs`; injected into `AppState` and referenced by +the auth extractor. + +## Graceful shutdown + +Hook into the existing SIGTERM handler in `main.rs` to call +`tracker.flush_now().await` before the runtime exits. Ensures rolling +restarts / container replacements don't lose the last 30 s of +liveness data. + +## Admin dashboard integration (deferred, sibling PR) + +Once `last_seen_at` exists, the admin sessions panel can render a +"last seen X min ago" column instead of only "created X ago". Small +follow-up — not part of this plan's scope, but the schema addition +unlocks it. + +## Testing + +Two hermetic units (no DB): + +1. `DashMap` dedup test — insert same key 10× with different + timestamps, drain, assert one entry with the latest timestamp. +2. `flush_now()` UPDATE-shape test — mock a `PgExecutor`, assert the + batched UNNEST binds match the drained set. Uses `sqlx-mock` or + equivalent. + +One integration (real PG, gated on `integration_tests` cfg): + +3. End-to-end — insert a session row, `stamp()` it, call `flush_now`, + assert `last_seen_at > created_at`. Covers the whole write path + including the `greatest()` guard. + +## Phasing + +1. **Migration** — add column + partial index. Ships alone; zero + application-layer impact. Reversible via `DROP COLUMN`. ✅ **2026-08-18** + (`migrations/20261014000000_sessions_last_seen_at.sql`). +2. **`LastSeenTracker` service** — DashMap + flusher task. Wire into + `AppState`. Middleware calls `stamp()`. Now `last_seen_at` moves + in real time. ✅ **2026-08-18** (`src/infrastructure/services/last_seen_tracker.rs`). +3. **JWT `sid` claim** — token minters carry the fresh session's + id; auth middleware reads it and stamps with no DB round trip. + New tokens carry it, old tokens still validate (Option → no-op). + ✅ **2026-08-18** (extends `TokenClaims::sid` on + `application/ports/auth_ports.rs`). +4. **Prometheus gauges** — background poller updates the three + gauges every 30 s using the queries above. Gated on + `OXICLOUD_METRICS_LISTEN` (no recorder → no periodic PG hits). + ✅ **2026-08-18** (`src/infrastructure/services/session_liveness_gauges.rs`). +5. **Graceful-shutdown flush** — hook into SIGTERM handler. + ✅ **2026-08-18** (added `shutdown_signal` + + `with_graceful_shutdown` in `main.rs`). +6. **Session DTO exposes `last_seen_at` + `is_online`** — + `GET /api/admin/sessions` returns both so the admin table + external + monitors can read them. `is_online` is the server-side derivation + against `ONLINE_WINDOW` (see terminology decision above). ✅ + **2026-08-18** (`application/dtos/session_dto.rs`). +7. **Admin dashboard "Online" column** (deferred to a sibling PR) — + render `is_online` as a green/grey dot next to each row plus + "last seen X ago" from `last_seen_at`. + +## Future — multi-instance + +If OxiCloud grows a multi-instance deployment story (K8s replicaset, +active-active behind a load balancer), the in-process DashMap becomes +insufficient — each process holds its own map, N flushers race +UPDATEs, coalescing across instances doesn't happen. + +Migration path when that becomes real: + +1. Keep the schema (`last_seen_at` column + partial index). +2. Replace the `LastSeenTracker::stamp()` in-process insert with a + `SELECT pg_notify('oxicloud_session_seen', $session_id)` call. +3. Move the flusher into a **single elected worker** (leader election + via advisory lock in PG). That worker `LISTEN`s the channel, + accumulates into an in-process HashMap, flushes on the same 30 s + ticker. +4. Every OxiCloud instance publishes; one instance consumes. + +Trade-off: NOTIFY costs a PG round-trip per request (~1 ms) vs the +current ~50 ns DashMap insert. Only pay that when multi-instance +coalescing actually matters. Schema and gauge queries stay identical; +only the tracker implementation swaps. + +## Open questions + +1. ~~Definition of "active"~~ — DECIDED 2026-08-18: hardcoded to + 5 min, no env var. See [Config surface](#config-surface). +2. **Should the gauge query drop long-inactive sessions?** — a + Nextcloud desktop client checking in every 6 h qualifies as + "active" if the window is 6 h. Probably want two separate metrics + (web-active < 5 min AND dav-active < 6 h) but scope creep for the + initial ship. +3. **Multi-tab dedup on the UI side** — three browser tabs of the + same user share ONE session (same cookies, same refresh token, + same row). Naturally deduped at the DB layer — no action needed. +4. **App-password rows** — Nextcloud desktop / mobile clients use + `auth.app_passwords` on top of the session model. Whether they + should get their own `last_seen_at` (and a companion metric) is + deferred; separate concern, separate PR. diff --git a/migrations/20261014000000_sessions_last_seen_at.sql b/migrations/20261014000000_sessions_last_seen_at.sql new file mode 100644 index 00000000..40b36c1d --- /dev/null +++ b/migrations/20261014000000_sessions_last_seen_at.sql @@ -0,0 +1,28 @@ +-- Session liveness tracking — per-request `last_seen_at` stamp on +-- `auth.sessions`, moved by the in-process `LastSeenTracker` +-- (see `src/application/services/last_seen_tracker.rs`) via a +-- batched UPDATE every 30 s. +-- +-- Distinct from `created_at`: that column moves on session ROTATION +-- (every silent refresh), so its resolution is capped at the +-- access-token TTL (default 3600 s). `last_seen_at` moves on every +-- authenticated request, so the "active in the last N min" query +-- underlying `oxicloud_sessions_active` / `_active_users` gauges is +-- accurate to the flusher's 30 s cadence regardless of token TTL. +-- +-- See `docs/plan/sessions.md` for the full design (why DashMap + +-- periodic flush, why partial index, why `NOW()` default). + +ALTER TABLE auth.sessions + ADD COLUMN last_seen_at TIMESTAMPTZ NOT NULL DEFAULT NOW(); + +-- Partial index — the only reads on this column are the gauge +-- queries in `session_liveness_gauges.rs`, and they always filter +-- `revoked = FALSE`. Indexing only unrevoked rows keeps the write +-- cost of the 30 s batched UPDATE flat: rotated / revoked rows +-- fall out of the index automatically when `revoked` flips to TRUE +-- (partial-index maintenance drops them, no re-scan). A full +-- b-tree on the column would double index size for zero read +-- benefit — every gauge query would skip the revoked half anyway. +CREATE INDEX idx_sessions_last_seen_at ON auth.sessions(last_seen_at) + WHERE revoked = FALSE; diff --git a/src/application/dtos/session_dto.rs b/src/application/dtos/session_dto.rs index 87fe23ba..170457ee 100644 --- a/src/application/dtos/session_dto.rs +++ b/src/application/dtos/session_dto.rs @@ -14,6 +14,8 @@ //! separate batch fetch (extra round-trip). Frontend cross-references //! `user_id` against its cached user list. +use std::time::Duration; + use chrono::{DateTime, Utc}; use serde::Serialize; use utoipa::ToSchema; @@ -21,6 +23,20 @@ use uuid::Uuid; use crate::domain::entities::session::{Session, SessionOrigin}; +/// The "recently seen" threshold that turns a session's +/// `last_seen_at` into a green-dot "Online" badge on the admin +/// sessions panel — AND the same window that drives the +/// `oxicloud_sessions_online[_users]` Prometheus gauges (see +/// `src/infrastructure/services/session_liveness_gauges.rs`). +/// The two MUST agree so the dashboard's per-row badge count +/// matches the gauge's aggregate — one source of truth here. +/// +/// 5 min feels responsive without over-fluctuating with +/// tab-open-then-close blips. Deliberately hardcoded, not an +/// env var — see `docs/plan/sessions.md` §"Config surface" for +/// the reasoning. +pub const ONLINE_WINDOW: Duration = Duration::from_secs(5 * 60); + /// Authenticated-caller context — the caller's identity + session- /// bound signals a service method might key off. Constructed at the /// handler boundary from `AuthUser` and passed through unchanged; @@ -53,6 +69,17 @@ pub struct SessionSummaryDto { pub user_id: Uuid, pub created_at: DateTime, pub expires_at: DateTime, + /// Wall-clock time this session was last observed serving an + /// authenticated request. Moved forward per request by the + /// in-process [`LastSeenTracker`](crate::infrastructure::services::last_seen_tracker) + /// via a batched UPDATE every 30 s — so this value trails the + /// true "last seen" by at most one flush interval on a running + /// server. On DB read it always converges after a graceful + /// shutdown flush. Distinct from `created_at`: that only moves + /// on session rotation (silent refresh), so its resolution is + /// capped at the access-token TTL. The admin table renders a + /// "last seen X ago" column off this field. + pub last_seen_at: DateTime, pub ip_address: Option, pub user_agent: Option, /// `true` iff the session is DPoP-bound. Rendered as a lock icon @@ -68,7 +95,24 @@ pub struct SessionSummaryDto { pub is_revoked: bool, /// Whether this row is currently usable — `!revoked && expires_at > now()`. /// Kept server-side so the SPA doesn't drift if the browser clock is off. + /// **Distinct from [`is_online`](Self::is_online)** — this is a + /// *lifecycle* signal (row still has authority), that one is a + /// *presence* signal (a request landed on it lately). pub is_active: bool, + /// Whether the session was actually observed serving a request in the + /// last [`ONLINE_WINDOW`] (5 min). Presence signal, orthogonal to + /// [`is_active`](Self::is_active): a session may be active-and-online + /// (green dot in the admin table), active-and-idle (no dot, "last + /// seen 12 min ago"), or non-active-and-offline (expired / revoked + /// rows are never online). Derived server-side against + /// [`ONLINE_WINDOW`] so the row-level badge stays consistent with + /// the `oxicloud_sessions_online[_users]` Prometheus aggregates. + /// + /// Guaranteed `false` for revoked / expired rows — those short- + /// circuit before the recency check so a revoked row that happened + /// to receive a request in its final second before revocation + /// doesn't confusingly render "Online" post-revocation. + pub is_online: bool, // NOTE: no `oidc_sid` / `oidc_sid_prefix` field. The IdP-emitted // sid identifies the row's upstream session and stays server-side // (used by Back-Channel Logout matching). Exposing even a prefix @@ -104,23 +148,40 @@ impl SessionSummaryDto { pub fn from_session(s: Session, caller_jkt: Option<&str>) -> Self { let is_revoked = s.is_revoked(); let is_expired = s.is_expired(); + let is_active = !is_revoked && !is_expired; let jkt = s.dpop_jkt().map(|s| s.to_owned()); let dpop_jkt_prefix = jkt.as_ref().map(|t| t.chars().take(8).collect::()); let is_current = match (jkt.as_deref(), caller_jkt) { (Some(row), Some(caller)) => row == caller, _ => false, }; + // Presence check gated on lifecycle — a revoked or expired + // row's `last_seen_at` may still be fresh (the last request + // that arrived just before revocation), but calling it + // "Online" post-revocation would confuse an admin reading + // the panel. Short-circuit on !is_active. + let online_cutoff = match chrono::Duration::from_std(ONLINE_WINDOW) { + Ok(d) => Utc::now() - d, + // Cast can only fail on a Duration too large for i64 + // milliseconds; not reachable with our 5 min constant. + // Fall back to "never online" rather than panic — a + // wrong badge is fixable, a request-path panic is not. + Err(_) => DateTime::::MAX_UTC, + }; + let is_online = is_active && s.last_seen_at() > online_cutoff; Self { id: s.id(), user_id: s.user_id(), created_at: s.created_at(), expires_at: s.expires_at(), + last_seen_at: s.last_seen_at(), ip_address: s.ip_address().map(str::to_owned), user_agent: s.user_agent().map(str::to_owned), is_bound: jkt.is_some(), dpop_jkt_prefix, is_revoked, - is_active: !is_revoked && !is_expired, + is_active, + is_online, origin: s.origin(), is_current, } @@ -166,6 +227,7 @@ mod tests { Some(sid.to_string()), None, crate::domain::entities::session::SessionOrigin::Oidc, + Utc::now(), ) } @@ -241,6 +303,121 @@ mod tests { assert!(dto.is_active); } + #[test] + fn dto_exposes_last_seen_at() { + // Regression: the admin table renders "last seen X ago" + // straight off this field, and clients that build + // dashboards off the session API rely on it too. Guards + // against a struct field being removed / renamed silently. + let s = base(false, None); + let expected = s.last_seen_at(); + let dto = SessionSummaryDto::from(s); + assert_eq!(dto.last_seen_at, expected); + let json = serde_json::to_string(&dto).unwrap(); + assert!( + json.contains("\"last_seen_at\""), + "wire shape must include `last_seen_at`: {json}" + ); + } + + /// A freshly-minted, unbound, unrevoked session ships with + /// `last_seen_at = Utc::now()` from `Session::new`, so it + /// MUST render as online. This is the green-dot happy path + /// the admin panel keys off — regression here means the + /// dashboard misses every currently-active session. + #[test] + fn dto_is_online_when_last_seen_is_fresh() { + let dto = SessionSummaryDto::from(base(false, None)); + assert!(dto.is_online, "fresh session must be online: {dto:?}"); + assert!(dto.is_active); + } + + /// A session whose `last_seen_at` is older than the + /// [`ONLINE_WINDOW`] MUST render as offline even when the + /// row is otherwise Active — that's the whole point of the + /// presence vs lifecycle split. Constructed via `from_raw` + /// so we can stamp a stale timestamp deterministically. + #[test] + fn dto_is_not_online_when_last_seen_is_stale() { + let stale = Utc::now() - chrono::Duration::hours(1); + let s = Session::from_raw( + Uuid::new_v4(), + Uuid::new_v4(), + "rt".to_string(), + Utc::now() + Duration::days(30), + None, + None, + stale, + false, + Uuid::new_v4(), + None, + None, + None, + SessionOrigin::Password, + stale, + ); + let dto = SessionSummaryDto::from(s); + assert!(!dto.is_online, "1h-idle session must not be online"); + assert!(dto.is_active, "stale-but-alive session stays active"); + } + + /// Anti-confusion guard: a revoked row whose `last_seen_at` + /// happens to be fresh (the last request that landed just + /// before revocation) must NOT surface as "Online" — an admin + /// reading the panel post-revocation expects the green dot + /// gone. `is_online` short-circuits on `!is_active`. + #[test] + fn dto_is_not_online_when_revoked_even_if_fresh() { + let dto = SessionSummaryDto::from(base(true, None)); + assert!(dto.is_revoked); + assert!(!dto.is_active); + assert!( + !dto.is_online, + "revoked-but-fresh row must never render as online", + ); + } + + /// Same anti-confusion guard for expiry: a session that's + /// past `expires_at` but whose last request landed in the + /// last 5 min must not surface as online. + #[test] + fn dto_is_not_online_when_expired_even_if_fresh() { + let past = Utc::now() - Duration::days(1); + let s = Session::from_raw( + Uuid::new_v4(), + Uuid::new_v4(), + "rt".to_string(), + past, // expires_at in the past + None, + None, + past, + false, + Uuid::new_v4(), + None, + None, + None, + SessionOrigin::Password, + Utc::now(), // last_seen_at fresh + ); + let dto = SessionSummaryDto::from(s); + assert!(!dto.is_active, "expired session is not active"); + assert!( + !dto.is_online, + "expired-but-fresh row must never render as online", + ); + } + + #[test] + fn fresh_session_has_last_seen_equal_to_created_at() { + // The DB default is `NOW()` and `Session::new` mirrors + // that with `Utc::now()` for BOTH columns — so a + // freshly-minted session immediately counts as "recently + // active" for the liveness gauges rather than showing up + // as long-idle for the first flush interval. + let s = base(false, None); + assert_eq!(s.created_at(), s.last_seen_at()); + } + #[test] fn from_raw_expired_session_is_not_active() { let past = Utc::now() - Duration::days(1); @@ -258,6 +435,7 @@ mod tests { None, None, crate::domain::entities::session::SessionOrigin::Unknown, + past, ); let dto = SessionSummaryDto::from(s); assert!(!dto.is_active); diff --git a/src/application/ports/auth_ports.rs b/src/application/ports/auth_ports.rs index cb8564fa..e7f2deff 100644 --- a/src/application/ports/auth_ports.rs +++ b/src/application/ports/auth_ports.rs @@ -63,6 +63,14 @@ pub struct TokenClaims { /// The DPoP middleware reads it from the already-validated token /// (no DB round trip) to enforce "bound session → proof required". pub dpop_jkt: Option, + /// Session identifier — the `auth.sessions.id` this access token + /// was minted for. Read by the auth middleware to stamp + /// per-session liveness via [`LastSeenTracker`](crate::infrastructure::services::last_seen_tracker) + /// with no DB round trip. `None` for tokens minted by builds + /// that predate the `sid` claim (backward compat during rollout; + /// harmless — the missing sid just means no stamp fires, and + /// the token still authenticates normally). + pub sid: Option, } /// Port for JWT token operations. @@ -81,6 +89,7 @@ pub trait TokenServicePort: Send + Sync + 'static { fn generate_access_token( &self, user: &User, + session_id: Option, dpop_jkt: Option<&str>, ) -> Result; diff --git a/src/application/services/auth_application_service.rs b/src/application/services/auth_application_service.rs index 6bd62e07..e042dcee 100644 --- a/src/application/services/auth_application_service.rs +++ b/src/application/services/auth_application_service.rs @@ -1269,21 +1269,17 @@ impl AuthApplicationService { None => None, }; - // Generate tokens using the injected token service. The - // access token carries the `cnf.jkt` binding when present, - // so the DPoP middleware can enforce "bound → proof required" - // straight from the already-validated JWT — no session-row - // lookup on the hot path. - let access_token = self - .token_service - .generate_access_token(&user, validated_jkt.as_deref())?; - - let refresh_token = self.token_service.generate_refresh_token(); - + // Construct the session FIRST so its id is available to the + // token mint below — the `sid` claim lets the auth middleware + // stamp per-session liveness with no DB round trip. Order + // was reversed as part of the `last_seen_at` wiring + // (`docs/plan/sessions.md`). + // // Save session — new login starts a new token family. DPoP // binding is set at INSERT time and immutable thereafter (see // `docs/plan/dpop.md` — a mutable bind would let an attacker // downgrade a bound session by re-binding to their own key). + let refresh_token = self.token_service.generate_refresh_token(); let mut session = Session::new( user.id(), refresh_token.clone(), @@ -1293,6 +1289,18 @@ impl AuthApplicationService { Uuid::new_v4(), origin, ); + + // Generate tokens using the injected token service. The + // access token carries the `cnf.jkt` binding when present, + // so the DPoP middleware can enforce "bound → proof required" + // straight from the already-validated JWT — no session-row + // lookup on the hot path. `sid` correlates the token to the + // session row constructed just above. + let access_token = self.token_service.generate_access_token( + &user, + Some(session.id()), + validated_jkt.as_deref(), + )?; if let Some(jkt) = validated_jkt { // Success-path audit — records the bind so operators can // correlate a session_id in the panel with the exact moment @@ -1562,8 +1570,9 @@ impl AuthApplicationService { // thread `dpop_jkt` into a GET body. Session is minted // unbound; the SPA calls `POST /api/auth/dpop/bind` // post-redirect to bind it (see Gate 3). Token accordingly - // ships without `cnf.jkt`. - let access_token = self.token_service.generate_access_token(&user, None)?; + // ships without `cnf.jkt`. Session constructed first so + // its id can feed the token's `sid` claim — see the login + // path above for the rationale. let refresh_token = self.token_service.generate_refresh_token(); let session = Session::new( user.id(), @@ -1574,6 +1583,9 @@ impl AuthApplicationService { Uuid::new_v4(), crate::domain::entities::session::SessionOrigin::MagicLink, ); + let access_token = + self.token_service + .generate_access_token(&user, Some(session.id()), None)?; self.session_storage.create_session(session).await?; tracing::info!( @@ -1715,16 +1727,12 @@ impl AuthApplicationService { )); } - // Generate new tokens. Inherit the DPoP binding from the - // parent session so the refreshed access token carries the - // same `cnf.jkt` — otherwise every refresh would silently - // downgrade to unbound and the next request would 401 under - // Gate 9 enforcement (see Gate 7). - let access_token = self - .token_service - .generate_access_token(&user, session.dpop_jkt())?; - let new_refresh_token = self.token_service.generate_refresh_token(); - + // Rotate the session first so the new row's id is available + // to the token mint below — the `sid` claim tracks the + // freshly-inserted row, not the revoked parent. Order + // reversed as part of the `last_seen_at` wiring + // (`docs/plan/sessions.md`). + // // New session inherits the family_id so reuse of any ancestor triggers // full-family revocation. Revoking the old session and inserting the // new one happen in ONE transaction (`rotate_session`) — this path @@ -1738,6 +1746,7 @@ impl AuthApplicationService { // refresh silently downgrade the session to unbound, and every // subsequent request would fail DPoP verification once required // mode enforces per-session binding. + let new_refresh_token = self.token_service.generate_refresh_token(); let mut new_session = Session::new( user.id(), new_refresh_token.clone(), @@ -1770,6 +1779,16 @@ impl AuthApplicationService { new_session = new_session.with_oidc_sid(sid.to_string()); } + // Mint the access token AFTER the new session is fully + // configured — the `sid` claim points at the new row's id, + // and `cnf.jkt` inherits from the parent so DPoP proof + // enforcement (Gate 9) still holds across the rotation. + let access_token = self.token_service.generate_access_token( + &user, + Some(new_session.id()), + session.dpop_jkt(), + )?; + self.session_storage .rotate_session(session.id(), new_session) .await?; @@ -4608,8 +4627,8 @@ impl AuthApplicationService { // through the browser's redirect chain. Session is minted // unbound; the SPA calls `POST /api/auth/dpop/bind` post- // redirect to bind it (see Gate 3). Token accordingly ships - // without `cnf.jkt`. - let access_token = self.token_service.generate_access_token(&user, None)?; + // without `cnf.jkt`. Session constructed first so its id + // feeds the token's `sid` claim. let refresh_token = self.token_service.generate_refresh_token(); let mut session = Session::new( @@ -4630,6 +4649,11 @@ impl AuthApplicationService { if let Some(sid) = claims.sid.as_ref() { session = session.with_oidc_sid(sid.clone()); } + // Mint AFTER session is fully configured so `sid` claim + // aligns with the row about to be inserted. + let access_token = + self.token_service + .generate_access_token(&user, Some(session.id()), None)?; self.session_storage.create_session(session).await?; let force_password_change = self.read_force_password_change(user.id()).await; diff --git a/src/application/services/device_auth_service.rs b/src/application/services/device_auth_service.rs index 8e0a455e..8e10751e 100644 --- a/src/application/services/device_auth_service.rs +++ b/src/application/services/device_auth_service.rs @@ -180,10 +180,12 @@ impl DeviceAuthService { // clients that don't run WebCrypto — always unbound (`None` // for the `dpop_jkt` param), which the DPoP middleware exempts // from proof requirements. See `docs/plan/dpop.md` Gate 9. - let access_token = self.token_service.generate_access_token(&user, None)?; + // + // Session constructed first so its id feeds the token's + // `sid` claim — the auth middleware uses this to stamp + // per-session liveness without a DB round trip + // (`docs/plan/sessions.md`). let refresh_token = self.token_service.generate_refresh_token(); - - // Persist refresh token as a session let session = Session::new( user_id, refresh_token.clone(), @@ -193,6 +195,9 @@ impl DeviceAuthService { Uuid::new_v4(), crate::domain::entities::session::SessionOrigin::Device, ); + let access_token = + self.token_service + .generate_access_token(&user, Some(session.id()), None)?; self.session_storage.create_session(session).await?; // Store tokens on the device code entity diff --git a/src/common/di.rs b/src/common/di.rs index 5a6fb9f8..9f3fc5b7 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -2126,6 +2126,38 @@ impl AppServiceFactory { let mut core = core; core.zip_service = Some(zip_service); + // Session liveness tracker — spawns its own 30 s flush + // loop at construction. Only built when auth (and thus + // sessions) exist; when auth is off this is `None` and + // the middleware never calls it. Uses the maintenance + // pool so background flushes don't compete with + // request-serving connections. See + // [`LastSeenTracker`](crate::infrastructure::services::last_seen_tracker). + let last_seen_tracker = if auth_services.is_some() { + Some( + crate::infrastructure::services::last_seen_tracker::LastSeenTracker::start( + maintenance_pool.clone(), + ), + ) + } else { + None + }; + + // Session-liveness Prometheus poller — three COUNT(*) reads + // every 30 s, publishing `oxicloud_sessions_active`, + // `_active_users`, `_total_non_revoked`. Only spawned when + // the Prometheus recorder is installed (i.e., + // `OXICLOUD_METRICS_LISTEN` is set) — without the recorder + // the `metrics::gauge!(...)` calls are no-ops and the + // periodic PG hits would be pure waste. Requires auth for + // the same reason as `last_seen_tracker`: no sessions to + // count without it. + if auth_services.is_some() && self.config.metrics_listen.is_some() { + crate::infrastructure::services::session_liveness_gauges::spawn( + maintenance_pool.clone(), + ); + } + // 9. Assemble final AppState let mut app_state = AppState { core, @@ -2158,6 +2190,7 @@ impl AppServiceFactory { people_service, storage_usage_service, grant_cleanup_service, + last_seen_tracker, calendar_service: None, calendar_use_case: None, addressbook_use_case: None, @@ -2952,6 +2985,16 @@ pub struct AppState { pub grant_cleanup_service: Option< Arc, >, + /// Per-session liveness tracker — the auth middleware calls + /// `stamp(session_id)` after every successful token validation, + /// and a background loop flushes the DashMap to `auth.sessions. + /// last_seen_at` every 30 s (batched UNNEST UPDATE). `None` + /// when auth is disabled — nothing to track. See + /// [`LastSeenTracker`](crate::infrastructure::services::last_seen_tracker) + /// for the contract and `docs/plan/sessions.md` for the design. + pub last_seen_tracker: Option< + Arc, + >, pub calendar_service: Option>, pub calendar_use_case: Option>, pub addressbook_use_case: Option>, diff --git a/src/domain/entities/session.rs b/src/domain/entities/session.rs index 7e30653a..cf2b10aa 100644 --- a/src/domain/entities/session.rs +++ b/src/domain/entities/session.rs @@ -97,6 +97,16 @@ pub struct Session { /// construction so a callsite can't forget to record it (the /// admin sessions panel filters on this). origin: SessionOrigin, + /// Wall-clock time the session was last observed serving an + /// authenticated request. Set to `created_at` at construction so + /// a freshly-minted session immediately counts as "recently + /// active" for the liveness gauges; moved forward in batches by + /// [`LastSeenTracker`](crate::infrastructure::services::last_seen_tracker) + /// via a per-30 s UNNEST-based UPDATE, so per-request writes + /// stay in-process. Distinct from `created_at` — that only moves + /// on session rotation (silent refresh), so its resolution is + /// capped at the access-token TTL. See `docs/plan/sessions.md`. + last_seen_at: DateTime, } impl Session { @@ -129,6 +139,7 @@ impl Session { oidc_sid: None, dpop_jkt: None, origin, + last_seen_at: now, } } @@ -180,6 +191,7 @@ impl Session { oidc_sid: Option, dpop_jkt: Option, origin: SessionOrigin, + last_seen_at: DateTime, ) -> Self { Self { id, @@ -195,6 +207,7 @@ impl Session { oidc_sid, dpop_jkt, origin, + last_seen_at, } } @@ -258,6 +271,10 @@ impl Session { pub fn origin(&self) -> SessionOrigin { self.origin } + + pub fn last_seen_at(&self) -> DateTime { + self.last_seen_at + } } #[cfg(test)] @@ -311,6 +328,7 @@ mod tests { None, Some("thumbprint-xyz".to_string()), SessionOrigin::Unknown, + Utc::now(), ); assert_eq!(s.dpop_jkt(), Some("thumbprint-xyz")); } diff --git a/src/infrastructure/repositories/pg/session_pg_repository.rs b/src/infrastructure/repositories/pg/session_pg_repository.rs index b4c638be..336b8e80 100644 --- a/src/infrastructure/repositories/pg/session_pg_repository.rs +++ b/src/infrastructure/repositories/pg/session_pg_repository.rs @@ -117,7 +117,7 @@ impl SessionRepository for SessionPgRepository { SELECT id, user_id, refresh_token, expires_at, ip_address, user_agent, created_at, revoked, family_id, - oidc_id_token, oidc_sid, dpop_jkt, origin + oidc_id_token, oidc_sid, dpop_jkt, origin, last_seen_at FROM auth.sessions WHERE id = $1 "#, @@ -141,6 +141,7 @@ impl SessionRepository for SessionPgRepository { row.get("oidc_sid"), row.get("dpop_jkt"), crate::domain::entities::session::SessionOrigin::from_wire(row.get("origin")), + row.get("last_seen_at"), )) } @@ -155,7 +156,7 @@ impl SessionRepository for SessionPgRepository { SELECT id, user_id, refresh_token, expires_at, ip_address, user_agent, created_at, revoked, family_id, - oidc_id_token, oidc_sid, dpop_jkt, origin + oidc_id_token, oidc_sid, dpop_jkt, origin, last_seen_at FROM auth.sessions WHERE refresh_token = $1 "#, @@ -179,6 +180,7 @@ impl SessionRepository for SessionPgRepository { row.get("oidc_sid"), row.get("dpop_jkt"), crate::domain::entities::session::SessionOrigin::from_wire(row.get("origin")), + row.get("last_seen_at"), )) } @@ -192,7 +194,7 @@ impl SessionRepository for SessionPgRepository { SELECT id, user_id, refresh_token, expires_at, ip_address, user_agent, created_at, revoked, family_id, - oidc_id_token, oidc_sid, dpop_jkt, origin + oidc_id_token, oidc_sid, dpop_jkt, origin, last_seen_at FROM auth.sessions WHERE user_id = $1 ORDER BY created_at DESC @@ -220,6 +222,7 @@ impl SessionRepository for SessionPgRepository { row.get("oidc_sid"), row.get("dpop_jkt"), crate::domain::entities::session::SessionOrigin::from_wire(row.get("origin")), + row.get("last_seen_at"), ) }) .collect(); @@ -248,7 +251,7 @@ impl SessionRepository for SessionPgRepository { SELECT id, user_id, refresh_token, expires_at, ip_address, user_agent, created_at, revoked, family_id, - oidc_id_token, oidc_sid, dpop_jkt, origin + oidc_id_token, oidc_sid, dpop_jkt, origin, last_seen_at FROM auth.sessions WHERE ($1::uuid IS NULL OR user_id = $1) AND ($2 OR (revoked = false AND expires_at > NOW())) @@ -281,6 +284,7 @@ impl SessionRepository for SessionPgRepository { row.get("oidc_sid"), row.get("dpop_jkt"), crate::domain::entities::session::SessionOrigin::from_wire(row.get("origin")), + row.get("last_seen_at"), ) }) .collect(); diff --git a/src/infrastructure/services/jwt_service.rs b/src/infrastructure/services/jwt_service.rs index 9f8385a2..939a37f5 100644 --- a/src/infrastructure/services/jwt_service.rs +++ b/src/infrastructure/services/jwt_service.rs @@ -55,6 +55,17 @@ struct JwtClaims { /// Serialised as `{"cnf": {"jkt": "..."}}` to match RFC 9449. #[serde(skip_serializing_if = "Option::is_none")] pub cnf: Option, + /// OIDC-style `sid` claim (RFC 8417 §4.1) — carries the + /// `auth.sessions.id` this access token was minted for so the + /// auth middleware can stamp per-session liveness without a DB + /// round trip. `None` on tokens minted by pre-`sid` builds so + /// deserialisation stays backward-compatible during rollout. + /// Kept as `String` on the wire (Uuid parses at the port + /// boundary) so a malformed value fails at token-decode time + /// with a clear parse error instead of poisoning the field + /// silently. + #[serde(skip_serializing_if = "Option::is_none")] + pub sid: Option, } /// RFC 9449 §5 confirmation-key wrapper. Only the `jkt` member is @@ -72,6 +83,17 @@ impl From for TokenClaims { // signed always carries a UUID `sub`; nil is a safe sentinel the // middleware rejects. See benches/ROUND14.md §A3. let sub_id = uuid::Uuid::parse_str(&claims.sub).unwrap_or_else(|_| uuid::Uuid::nil()); + // Parse `sid` at the boundary — same amortization rationale + // as `sub_id` above, and gives us a clean `Option` in + // `TokenClaims`. A parse failure (mint-time bug or hand- + // crafted claim) drops the sid to `None`; the middleware + // then simply skips the stamp — token still authenticates. + // Legitimate tokens minted by this codebase always carry a + // valid Uuid, so this only masks external drift. + let sid = claims + .sid + .as_deref() + .and_then(|s| uuid::Uuid::parse_str(s).ok()); TokenClaims { sub_id, sub: claims.sub, @@ -82,6 +104,7 @@ impl From for TokenClaims { email: claims.email, role: claims.role, dpop_jkt: claims.cnf.map(|c| c.jkt), + sid, } } } @@ -196,6 +219,7 @@ impl TokenServicePort for JwtTokenService { fn generate_access_token( &self, user: &User, + session_id: Option, dpop_jkt: Option<&str>, ) -> Result { let now = Utc::now().timestamp(); @@ -219,6 +243,7 @@ impl TokenServicePort for JwtTokenService { cnf: dpop_jkt.map(|jkt| CnfClaim { jkt: jkt.to_string(), }), + sid: session_id.map(|id| id.to_string()), }; // Log JWT claims for debugging @@ -331,7 +356,7 @@ mod tests { let user = create_test_user(); let token = service - .generate_access_token(&user, None) + .generate_access_token(&user, Some(Uuid::new_v4()), None) .expect("Should generate token"); let claims = service @@ -370,7 +395,7 @@ mod tests { let user = create_test_user(); let token = service - .generate_access_token(&user, None) + .generate_access_token(&user, Some(Uuid::new_v4()), None) .expect("Should generate token"); // First call: cache miss — performs full HMAC verification @@ -397,7 +422,7 @@ mod tests { 86400, ); let token = service - .generate_access_token(&create_test_user(), None) + .generate_access_token(&create_test_user(), Some(Uuid::new_v4()), None) .expect("Should generate token"); // Miss populates the cache; hit must hand back the very same @@ -425,4 +450,46 @@ mod tests { let (hits, _misses) = service.cache_stats(); assert_eq!(hits, 0, "Invalid tokens should never produce cache hits"); } + + /// Regression for the `sid` claim wiring — the auth middleware + /// stamps per-session liveness by reading this exact field. If + /// the mint stops setting the claim or the port stops parsing + /// it, every `LastSeenTracker::stamp` call goes silent and the + /// Prometheus gauges freeze at zero. + #[test] + fn access_token_round_trips_session_id_as_sid_claim() { + let service = JwtTokenService::new( + "test_secret_key_at_least_32_bytes_long".to_string(), + 3600, + 86400, + ); + let user = create_test_user(); + let session_id = Uuid::new_v4(); + let token = service + .generate_access_token(&user, Some(session_id), None) + .expect("Should generate token"); + let claims = service.validate_token(&token).expect("Should validate"); + assert_eq!(claims.sid, Some(session_id)); + } + + /// Backward-compatibility guard: a mint call with `None` + /// omits the `sid` claim entirely (matches the pre-`sid` + /// on-wire shape), and the validated claims surface `None` + /// on the port. The middleware's `if let (Some(sid), ...)` + /// then simply skips the stamp — critical during rollout + /// where old tokens are still in flight. + #[test] + fn access_token_without_session_id_omits_sid_claim() { + let service = JwtTokenService::new( + "test_secret_key_at_least_32_bytes_long".to_string(), + 3600, + 86400, + ); + let user = create_test_user(); + let token = service + .generate_access_token(&user, None, None) + .expect("Should generate token"); + let claims = service.validate_token(&token).expect("Should validate"); + assert_eq!(claims.sid, None); + } } diff --git a/src/infrastructure/services/last_seen_tracker.rs b/src/infrastructure/services/last_seen_tracker.rs new file mode 100644 index 00000000..577b8a05 --- /dev/null +++ b/src/infrastructure/services/last_seen_tracker.rs @@ -0,0 +1,239 @@ +//! Per-session liveness tracker — the hot path of the "how many +//! sessions are active right now?" observation loop. +//! +//! **Contract.** Every authenticated request calls +//! [`LastSeenTracker::stamp`] with the session id it resolved. The +//! call is O(1) — a DashMap upsert of `(session_id → Utc::now())` — +//! and hits no I/O. The map data structure IS the dedup: 100 +//! requests against the same session in a flush window contribute +//! ONE row to the batched UPDATE with the latest timestamp. +//! +//! A background task ([`flush_loop`](Self::flush_loop), spawned at +//! construction) drains the map every 30 s and issues one +//! `UPDATE ... FROM UNNEST($1::uuid[], $2::timestamptz[])` covering +//! every distinct session_id observed in the window. The +//! `greatest(s.last_seen_at, t.seen_at)` guard makes the write +//! idempotent under any retry / race / clock skew — replaying the +//! same batch never moves the column backward. +//! +//! **Failure model.** A flush that hits a transient PG error does +//! NOT drop the accumulated set — the map is not cleared until the +//! UPDATE succeeds. Next tick overlays new activity on the retry +//! set and the whole thing gets flushed together. Bounded loss +//! window under a hard crash is one flush interval; graceful +//! shutdown calls [`flush_now`](Self::flush_now) synchronously (see +//! `main.rs`) so rolling restarts drop nothing. +//! +//! **Non-goals.** No per-session locking, no ordering guarantees +//! across sessions, no back-pressure on the flusher (the loop +//! swallows errors and keeps ticking). The workload is +//! observation-only — losing a stamp under contention is a +//! correctness no-op, the next request re-stamps. +//! +//! See `docs/plan/sessions.md` for the full design (why DashMap +//! over Mutex, why not NOTIFY/LISTEN today, migration +//! path to a multi-instance cluster). + +use std::sync::Arc; +use std::time::Duration; + +use chrono::{DateTime, Utc}; +use dashmap::DashMap; +use sqlx::PgPool; +use uuid::Uuid; + +/// Cadence of the batched UPDATE. Hardcoded — 30 s balances DB +/// write load against gauge freshness (the Prometheus scrape +/// interval is typically 15 s, so at worst two scrapes see the +/// same value before the next flush). Deliberately NOT exposed as +/// an env var — tuning it is a deployment-shape question we've +/// never had to answer in practice. +const FLUSH_INTERVAL: Duration = Duration::from_secs(30); + +/// In-process session-liveness tracker. See [module docs](self) for +/// the full contract; the two entry points are: +/// +/// - [`stamp`](Self::stamp) — called from the auth middleware on +/// every authenticated request. +/// - [`flush_now`](Self::flush_now) — called from the graceful- +/// shutdown handler. +/// +/// The periodic flush task is spawned on the tokio runtime by +/// [`start`](Self::start) at construction. The struct keeps no +/// handle to it — the task holds the `Arc` and observes the +/// runtime shutting down naturally. +pub struct LastSeenTracker { + /// (session_id → last observed time). DashMap's sharded locking + /// parallelises writes across distinct session_ids — different + /// users' requests never contend. + seen: DashMap>, + /// Maintenance pool — the tracker is a background writer and + /// must not compete with request-serving connections. + pool: Arc, +} + +impl LastSeenTracker { + /// Construct + spawn the flush loop. Returns the shared + /// handle; callers store it on `AppState` and pass it to the + /// auth middleware. + /// + /// The background task lives for the runtime's lifetime — no + /// cancellation handle is exposed because there is no + /// mid-process reason to stop tracking (a stopped flusher is + /// indistinguishable from a wedged one, and both are bugs). + /// Graceful shutdown calls [`flush_now`](Self::flush_now) + /// separately BEFORE the runtime tears down. + pub fn start(pool: Arc) -> Arc { + let this = Arc::new(Self { + seen: DashMap::new(), + pool, + }); + tokio::spawn(this.clone().flush_loop()); + this + } + + /// Record that `session_id` was observed serving a request + /// right now. Overwrites any prior stamp for the same session + /// in the current window — the flusher uses the latest value. + /// + /// O(1) DashMap upsert. No I/O. Never fails. + pub fn stamp(&self, session_id: Uuid) { + self.seen.insert(session_id, Utc::now()); + } + + /// Drain the accumulated stamps and write them in one batched + /// UPDATE. Idempotent — the `greatest(...)` guard means + /// replaying the same batch (or overlapping batches from a + /// retry) never moves the column backward. + /// + /// Errors are surfaced to the caller so `flush_loop`'s + /// warn-and-continue policy is a deliberate choice made in one + /// place, and the shutdown flusher in `main.rs` can decide + /// whether to log or panic. + /// + /// On PG error the accumulated set is NOT cleared — the next + /// tick retries with fresh activity overlaid. + pub async fn flush_now(&self) -> Result { + if self.seen.is_empty() { + return Ok(0); + } + + // Drain into two parallel vectors — one UNNEST arg each. + // `retain(|_,_| false)` clears every shard in-place; the + // pull-and-drop order doesn't matter (we upserted the + // latest wins per key already). + let mut ids: Vec = Vec::with_capacity(self.seen.len()); + let mut seen_at: Vec> = Vec::with_capacity(self.seen.len()); + for entry in self.seen.iter() { + ids.push(*entry.key()); + seen_at.push(*entry.value()); + } + + let result = sqlx::query( + r#" + UPDATE auth.sessions AS s + SET last_seen_at = greatest(s.last_seen_at, t.seen_at) + FROM UNNEST($1::uuid[], $2::timestamptz[]) AS t(id, seen_at) + WHERE s.id = t.id + "#, + ) + .bind(&ids) + .bind(&seen_at) + .execute(&*self.pool) + .await?; + + // Only clear the drained keys on success. A key inserted + // BETWEEN our copy above and the clear below survives + // (retain drops only those whose value we already flushed, + // by timestamp equality). Same-key re-stamp with a newer + // timestamp gets kept for the next flush. + let flushed: std::collections::HashMap> = + ids.iter().copied().zip(seen_at.iter().copied()).collect(); + self.seen + .retain(|k, v| flushed.get(k).is_none_or(|ts| ts != v)); + + let updated = result.rows_affected() as usize; + tracing::debug!( + target: "oxicloud::sessions", + batched = ids.len(), + updated, + "last_seen flush", + ); + Ok(updated) + } + + /// The periodic drain loop. Runs forever; every failed flush + /// is logged at WARN and the accumulated set is preserved for + /// the next tick. + async fn flush_loop(self: Arc) { + let mut ticker = tokio::time::interval(FLUSH_INTERVAL); + // Skip the "first tick fires immediately" behaviour — the + // map is empty at spawn time, so a same-tick flush is + // wasted work. + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + ticker.tick().await; + + loop { + ticker.tick().await; + if let Err(err) = self.flush_now().await { + tracing::warn!( + target: "oxicloud::sessions", + error = %err, + "last_seen flush failed; will retry next tick", + ); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Ten stamps of the same session_id must collapse to ONE + /// entry with the newest timestamp — the whole point of the + /// DashMap-as-dedup pattern. Guards against a future refactor + /// that swaps to an append-only channel and doubles the DB + /// write rate. + #[test] + fn stamps_dedup_by_session_id() { + // No pool needed — we're only exercising the map. Build + // the tracker directly without spawning the loop. + let seen = DashMap::new(); + let session = Uuid::new_v4(); + + for _ in 0..10 { + seen.insert(session, Utc::now()); + } + + assert_eq!(seen.len(), 1); + } + + /// Latest-wins semantics: two stamps for the same session + /// leave the newer timestamp in place, matching the flusher's + /// `greatest(...)` guard so a request that beats the flush + /// keeps its more recent stamp. + #[test] + fn stamp_keeps_latest_timestamp() { + let seen: DashMap> = DashMap::new(); + let session = Uuid::new_v4(); + + let t1 = Utc::now(); + seen.insert(session, t1); + let t2 = t1 + chrono::Duration::seconds(5); + seen.insert(session, t2); + + assert_eq!(*seen.get(&session).unwrap(), t2); + } + + /// Distinct sessions never collide — sharded map, no dedup + /// across keys. + #[test] + fn different_sessions_are_independent() { + let seen: DashMap> = DashMap::new(); + for _ in 0..100 { + seen.insert(Uuid::new_v4(), Utc::now()); + } + assert_eq!(seen.len(), 100); + } +} diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index b86e7560..7eeab9c5 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -27,6 +27,7 @@ pub mod folders_consistency_service; pub mod grant_cleanup_service; pub mod image_transcode_service; pub mod jwt_service; +pub mod last_seen_tracker; pub mod local_blob_backend; pub mod local_fs_mount_provider; pub mod login_lockout_service; @@ -51,6 +52,7 @@ pub mod retry_blob_backend; pub mod s3_blob_backend; pub mod search_index; pub mod session_cleanup_service; +pub mod session_liveness_gauges; pub mod share_unlock_cookie; pub mod smtp_email_sender; pub mod swappable_blob_backend; diff --git a/src/infrastructure/services/session_liveness_gauges.rs b/src/infrastructure/services/session_liveness_gauges.rs new file mode 100644 index 00000000..cc6f0cfe --- /dev/null +++ b/src/infrastructure/services/session_liveness_gauges.rs @@ -0,0 +1,169 @@ +//! Prometheus session-liveness gauges — periodic polling of +//! `auth.sessions` to publish three gauges the `/metrics` scraper +//! reads: +//! +//! - `oxicloud_sessions_online` — non-revoked rows observed in the +//! last [`ONLINE_WINDOW`](crate::application::dtos::session_dto::ONLINE_WINDOW). +//! **Per-session count**, not per-user — one user with three +//! devices contributes three. +//! - `oxicloud_sessions_online_users` — DISTINCT `user_id` behind +//! those online sessions. The multi-device factor is exactly +//! `sessions_online / sessions_online_users`. +//! - `oxicloud_sessions_total_non_revoked` — long-tail total, +//! including mobile clients still holding a refresh token they +//! haven't used in weeks. Useful sanity signal on the dashboard. +//! +//! **Naming — "online" vs "active".** The word "active" is already +//! spoken for by the session *lifecycle* (Active | Expired | +//! Revoked in the admin panel). Presence (recently-seen) is +//! orthogonal and uses "online" throughout the UI, DTO +//! (`SessionSummaryDto::is_online`), and these gauges — so a +//! dashboard graph and a per-row green-dot badge have the same +//! label root. Terminology decided 2026-08-18; see +//! `docs/plan/sessions.md`. +//! +//! **Cadence.** Poller ticks every [`POLL_INTERVAL`] (30 s). Three +//! `COUNT(*)` reads on the maintenance pool per tick — negligible +//! load on tens-of-thousands-of-rows tables thanks to the partial +//! index `idx_sessions_last_seen_at` (partial on `revoked = FALSE`, +//! which every query below filters on). +//! +//! **When it runs.** Spawned from DI only when auth is enabled AND +//! `OXICLOUD_METRICS_LISTEN` is set (recorder installed). Without +//! the recorder, `metrics::gauge!(...)` is a no-op — spawning +//! anyway would still hit PG every 30 s for values nobody reads. +//! +//! See `docs/plan/sessions.md` for the full design. + +use std::sync::Arc; +use std::time::Duration; + +use sqlx::PgPool; + +use crate::application::dtos::session_dto::ONLINE_WINDOW; + +/// Poll cadence. Matches the [`LastSeenTracker`](super::last_seen_tracker) +/// flush cadence so the gauges converge one tick after the tracker +/// flushes — no need to sync the two. +const POLL_INTERVAL: Duration = Duration::from_secs(30); + +/// Spawn the session-liveness poller. Detached — the task lives +/// for the runtime's lifetime; there's no mid-process reason to +/// stop reporting gauges. +/// +/// Emits an initial poll on spawn so the very first `/metrics` +/// scrape after boot returns real values instead of the recorder's +/// zero-initialised default. +pub fn spawn(maintenance_pool: Arc) { + tokio::spawn(async move { + // Immediate first tick — a scraper hitting `/metrics` in + // the first 30 s otherwise sees `oxicloud_sessions_online + // 0` even on a busy server. Warmup query is cheap. + if let Err(err) = poll_once(&maintenance_pool).await { + tracing::warn!( + target: "oxicloud::sessions", + error = %err, + "initial session-liveness poll failed", + ); + } + + let mut ticker = tokio::time::interval(POLL_INTERVAL); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + // Consume the first tick — `interval` fires immediately on + // creation and we've already done the warmup above. + ticker.tick().await; + + loop { + ticker.tick().await; + if let Err(err) = poll_once(&maintenance_pool).await { + tracing::warn!( + target: "oxicloud::sessions", + error = %err, + "session-liveness poll failed; keeping last-known gauge values", + ); + } + } + }); + tracing::info!( + target: "oxicloud::sessions", + poll_interval_secs = POLL_INTERVAL.as_secs(), + online_window_secs = ONLINE_WINDOW.as_secs(), + "📊 session-liveness gauges spawned", + ); +} + +/// One poll cycle. Three lightweight `COUNT` reads → three gauge +/// updates. Errors propagate to the caller (loop logs + retries +/// next tick; gauges keep their last-known value in the interim, +/// which is the honest thing to publish — a temporary PG blip is +/// not a "sessions dropped to zero" event). +async fn poll_once(pool: &PgPool) -> Result<(), sqlx::Error> { + // NOTE: `ONLINE_WINDOW` is a Duration; PG expects the interval + // in seconds via `make_interval` (portable across sqlx driver + // versions). Casting once at bind time is cheaper than an + // `INTERVAL '$1 seconds'` string interp and keeps the query + // parameterised. + let online_secs: f64 = ONLINE_WINDOW.as_secs_f64(); + + let online: i64 = sqlx::query_scalar( + r#" + SELECT COUNT(*) FROM auth.sessions + WHERE revoked = FALSE + AND last_seen_at > NOW() - make_interval(secs => $1) + "#, + ) + .bind(online_secs) + .fetch_one(pool) + .await?; + + let online_users: i64 = sqlx::query_scalar( + r#" + SELECT COUNT(DISTINCT user_id) FROM auth.sessions + WHERE revoked = FALSE + AND last_seen_at > NOW() - make_interval(secs => $1) + "#, + ) + .bind(online_secs) + .fetch_one(pool) + .await?; + + let total_non_revoked: i64 = sqlx::query_scalar( + r#" + SELECT COUNT(*) FROM auth.sessions WHERE revoked = FALSE + "#, + ) + .fetch_one(pool) + .await?; + + // metrics-exporter-prometheus takes f64 gauges; the raw COUNT + // fits into f64 precisely up to 2^53, well past any realistic + // session-row count. `describe_gauge!` is called once at first + // emission and cached in the recorder — the second/third tick + // just updates the value. + metrics::describe_gauge!( + "oxicloud_sessions_online", + "Non-revoked sessions observed in the last ONLINE_WINDOW." + ); + metrics::gauge!("oxicloud_sessions_online").set(online as f64); + + metrics::describe_gauge!( + "oxicloud_sessions_online_users", + "Distinct users behind sessions observed in the last ONLINE_WINDOW." + ); + metrics::gauge!("oxicloud_sessions_online_users").set(online_users as f64); + + metrics::describe_gauge!( + "oxicloud_sessions_total_non_revoked", + "Total non-revoked sessions regardless of last-seen recency." + ); + metrics::gauge!("oxicloud_sessions_total_non_revoked").set(total_non_revoked as f64); + + tracing::debug!( + target: "oxicloud::sessions", + online, + online_users, + total_non_revoked, + "session-liveness gauges updated", + ); + Ok(()) +} diff --git a/src/interfaces/middleware/auth.rs b/src/interfaces/middleware/auth.rs index 7f2a70ec..a2f2f6a1 100644 --- a/src/interfaces/middleware/auth.rs +++ b/src/interfaces/middleware/auth.rs @@ -229,6 +229,18 @@ pub async fn auth_middleware( request.extensions_mut().insert(current_user); tracing::Span::current() .record("user_id", tracing::field::display(user_id)); + // Bump per-session liveness for the + // Prometheus gauges. O(1) DashMap upsert + // — no I/O on this hot path. The `sid` + // claim is `None` on tokens minted by + // pre-`sid` builds, in which case the + // stamp is skipped entirely — no + // fallback lookup, no round-trip. + if let (Some(sid), Some(tracker)) = + (claims.sid, state.last_seen_tracker.as_ref()) + { + tracker.stamp(sid); + } return Ok(next.run(request).await); } Err(e) => { @@ -353,6 +365,15 @@ pub async fn auth_middleware( request.extensions_mut().insert(CookieAuthenticated); tracing::Span::current() .record("user_id", tracing::field::display(user_id)); + // Cookie-auth branch stamps the same + // way as the Bearer branch above — + // see that site for the O(1) / + // no-DB rationale. + if let (Some(sid), Some(tracker)) = + (claims.sid, state.last_seen_tracker.as_ref()) + { + tracker.stamp(sid); + } return Ok(next.run(request).await); } LiveRole::Revoked => { diff --git a/src/main.rs b/src/main.rs index c25fcc0c..7d936460 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1409,17 +1409,89 @@ async fn run() -> Result<(), Box> { let listener = tokio::net::TcpListener::from_std(socket.into())?; + // Grab shutdown-hook handles BEFORE the router consumes + // app_state below. Currently: only the session `LastSeenTracker` + // needs a final synchronous flush on graceful shutdown so + // rolling restarts don't lose the last flush interval of + // liveness stamps. Future services with shutdown obligations + // add their handles here and chain their flushes into + // [`shutdown_signal`] alongside this one. + let last_seen_tracker = app_state.last_seen_tracker.clone(); + // Provide the fully-built state to the router let app = app.with_state(app_state); // TCP_NODELAY is inherited from the listening socket on Linux, // so every accepted connection already has Nagle disabled. + // + // `with_graceful_shutdown` waits for SIGTERM / SIGINT, then + // stops accepting new connections, drains in-flight requests, + // and runs the async block below. The session tracker flush + // fires AFTER draining so it captures any last-second requests + // that landed while shutdown propagates. axum::serve( listener, app.into_make_service_with_connect_info::(), ) + .with_graceful_shutdown(async move { + shutdown_signal().await; + if let Some(tracker) = last_seen_tracker { + match tracker.flush_now().await { + Ok(n) => tracing::info!( + target: "oxicloud::sessions", + flushed = n, + "last-seen final flush before shutdown", + ), + Err(err) => tracing::warn!( + target: "oxicloud::sessions", + error = %err, + "last-seen final flush failed; up to one flush interval of \ + liveness data may have been lost", + ), + } + } + }) .await?; tracing::info!("Server shutdown completed"); Ok(()) } + +/// Block until the process receives SIGINT (Ctrl-C) or SIGTERM +/// (systemd / docker stop / K8s pod eviction). Returns once EITHER +/// arrives — no distinction between them at the caller: a signal +/// is a signal, drain and exit. +/// +/// On non-Unix (Windows), the `terminate` arm is a never-resolving +/// future so only Ctrl-C works — which matches how Windows expects +/// service shutdown to be signalled anyway. +async fn shutdown_signal() { + let ctrl_c = async { + if let Err(err) = tokio::signal::ctrl_c().await { + tracing::warn!("failed to install Ctrl-C handler: {err}"); + } + }; + + #[cfg(unix)] + let terminate = async { + match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) { + Ok(mut s) => { + s.recv().await; + } + Err(err) => { + tracing::warn!("failed to install SIGTERM handler: {err}"); + // Fall through to a pending future so tokio::select! doesn't + // spin — Ctrl-C is still armed. + std::future::pending::<()>().await; + } + } + }; + + #[cfg(not(unix))] + let terminate = std::future::pending::<()>(); + + tokio::select! { + _ = ctrl_c => tracing::info!("SIGINT received, initiating graceful shutdown"), + _ = terminate => tracing::info!("SIGTERM received, initiating graceful shutdown"), + } +}