From e7563d2eb754e7e42e08470405a3b5ececa751cd Mon Sep 17 00:00:00 2001 From: uddhavdave Date: Mon, 26 Oct 2026 17:18:36 +0000 Subject: [PATCH] refactor: read_org_setting and a shared TtlCache own the cached reads (#15840) --- .../src/request/alerts/incidents.rs | 7 +- src/core/src/logs/mod.rs | 2 +- src/core/src/self_reporting/ingestion.rs | 3 +- src/db/src/organization.rs | 127 +++--------------- src/infra/src/table/cipher.rs | 42 +++--- src/infra/src/table/mod.rs | 1 + src/infra/src/table/oncall_ownership.rs | 24 ++-- src/infra/src/table/oncall_policies.rs | 39 +++--- src/infra/src/table/oncall_schedules.rs | 22 ++- src/infra/src/table/oncall_teams.rs | 42 +++--- src/infra/src/table/org_ingestion_tokens.rs | 24 ++-- src/infra/src/table/synthetics_agents.rs | 22 ++- src/infra/src/table/synthetics_checks.rs | 30 ++--- .../src/table/synthetics_probe_tokens.rs | 35 ++--- src/infra/src/table/synthetics_variables.rs | 26 ++-- src/infra/src/table/ttl_cache.rs | 93 +++++++++++++ src/jobs/src/job/red_insights.rs | 21 +-- 17 files changed, 260 insertions(+), 300 deletions(-) create mode 100644 src/infra/src/table/ttl_cache.rs diff --git a/src/api/management/src/request/alerts/incidents.rs b/src/api/management/src/request/alerts/incidents.rs index 76f29ddae6..683f225367 100644 --- a/src/api/management/src/request/alerts/incidents.rs +++ b/src/api/management/src/request/alerts/incidents.rs @@ -960,7 +960,8 @@ pub async fn get_incident_rca_thread( }; let show_tool_io = match &session { RunSession::Present(_) => { - match db::organization::get_org_setting_ai_rca_thread_tool_output(&org_id).await { + match db::organization::read_org_setting(&org_id, |s| s.ai_rca_thread_tool_output).await + { Ok(show) => show, Err(e) => return MetaHttpResponse::internal_error(e), } @@ -1566,7 +1567,9 @@ pub async fn get_incident_rca_live_stream( return response; } let show_tool_io = - match db::organization::get_org_setting_ai_rca_thread_tool_output(&org_id).await { + match db::organization::read_org_setting(&org_id, |s| s.ai_rca_thread_tool_output) + .await + { Ok(show) => show, Err(e) => return MetaHttpResponse::internal_error(e), }; diff --git a/src/core/src/logs/mod.rs b/src/core/src/logs/mod.rs index 53e5f84519..fd9d6988a1 100644 --- a/src/core/src/logs/mod.rs +++ b/src/core/src/logs/mod.rs @@ -657,7 +657,7 @@ async fn ingestion_log_enabled() -> bool { return false; } // the logging will be enabled through meta only - db::organization::get_org_setting_toggle_ingestion_logs(META_ORG_ID) + db::organization::read_org_setting(META_ORG_ID, |s| s.toggle_ingestion_logs) .await .unwrap_or(false) } diff --git a/src/core/src/self_reporting/ingestion.rs b/src/core/src/self_reporting/ingestion.rs index ac80bd6e5d..68e4080547 100644 --- a/src/core/src/self_reporting/ingestion.rs +++ b/src/core/src/self_reporting/ingestion.rs @@ -241,7 +241,8 @@ pub(super) async fn ingest_usages(mut curr_usages: Vec) { // do not skip self-reporting even if super org reporting fails // Check if org has usage stream enabled - match crate::db::organization::get_org_setting_usage_stream_enabled(&org).await { + match crate::db::organization::read_org_setting(&org, |s| s.usage_stream_enabled).await + { Ok(true) => {} Ok(false) => continue, // self-report not enabled, so skip Err(e) => { diff --git a/src/db/src/organization.rs b/src/db/src/organization.rs index d0a3c480d6..07869042ae 100644 --- a/src/db/src/organization.rs +++ b/src/db/src/organization.rs @@ -87,29 +87,31 @@ pub async fn get_org_setting(org_id: &str) -> Result } }; + let mut settings = read_org_setting(org_id, OrganizationSetting::clone).await?; + settings.free_trial_expiry = trial_period_expiry; + Ok(settings) +} + +/// Picks from the org's cached settings, caching the stored or default settings on a miss. +pub async fn read_org_setting( + org_id: &str, + pick: impl FnOnce(&OrganizationSetting) -> T, +) -> Result { let key = format!("{ORG_SETTINGS_KEY_PREFIX}/{org_id}"); if let Some(v) = ORGANIZATION_SETTING.read().await.get(&key) { - let mut ret = v.clone(); - ret.free_trial_expiry = trial_period_expiry; - return Ok(ret); + return Ok(pick(v)); } - // Try to get settings from DB, but use default if not found - let mut settings: OrganizationSetting = match db::get(&key).await { + let settings: OrganizationSetting = match db::get(&key).await { Ok(settings) => json::from_slice(&settings)?, Err(Error::DbError(infra::errors::DbError::KeyNotExists(_))) => { OrganizationSetting::default() } Err(e) => return Err(e), }; - - // Cache the org setting (even if it's default) - ORGANIZATION_SETTING - .write() - .await - .insert(key.to_string(), settings.clone()); - settings.free_trial_expiry = trial_period_expiry; - Ok(settings) + let picked = pick(&settings); + ORGANIZATION_SETTING.write().await.insert(key, settings); + Ok(picked) } /// Applies `apply` to the settings in the store's `get_for_update`; true if it wrote them. @@ -120,105 +122,6 @@ where update_org_setting_in(infra::db::get_db().await.as_ref(), org_id, apply).await } -/// Get the toggle ingestion logs setting for an org -/// If the setting is not found, return false -/// we add a separate function for avoid clone the setting -pub async fn get_org_setting_toggle_ingestion_logs(org_id: &str) -> Result { - let key = format!("{ORG_SETTINGS_KEY_PREFIX}/{org_id}"); - if let Some(v) = ORGANIZATION_SETTING.read().await.get(&key) { - return Ok(v.toggle_ingestion_logs); - } - - // Try to get settings from DB, but use default if not found - let settings: OrganizationSetting = match db::get(&key).await { - Ok(settings) => json::from_slice(&settings)?, - Err(Error::DbError(infra::errors::DbError::KeyNotExists(_))) => { - OrganizationSetting::default() - } - Err(e) => return Err(e), - }; - let toggle_ingestion_logs = settings.toggle_ingestion_logs; - - // Cache the org setting (even if it's default) - ORGANIZATION_SETTING - .write() - .await - .insert(key.to_string(), settings); - Ok(toggle_ingestion_logs) -} - -/// Get the usage stream enabled setting for an org. -/// Without a stored setting this falls back to `default_usage_stream_enabled`, which is off. -pub async fn get_org_setting_usage_stream_enabled(org_id: &str) -> Result { - let key = format!("{ORG_SETTINGS_KEY_PREFIX}/{org_id}"); - if let Some(v) = ORGANIZATION_SETTING.read().await.get(&key) { - return Ok(v.usage_stream_enabled); - } - - // Try to get settings from DB, but use default if not found - let settings: OrganizationSetting = match db::get(&key).await { - Ok(settings) => json::from_slice(&settings)?, - Err(Error::DbError(infra::errors::DbError::KeyNotExists(_))) => { - OrganizationSetting::default() - } - Err(e) => return Err(e), - }; - let usage_stream_enabled = settings.usage_stream_enabled; - - // Cache the org setting (even if it's default) - ORGANIZATION_SETTING - .write() - .await - .insert(key.to_string(), settings); - Ok(usage_stream_enabled) -} - -/// Whether automatic RED anomaly insights are on for an org; absent settings mean off. -pub async fn get_org_setting_red_insights_enabled(org_id: &str) -> Result { - let key = format!("{ORG_SETTINGS_KEY_PREFIX}/{org_id}"); - if let Some(v) = ORGANIZATION_SETTING.read().await.get(&key) { - return Ok(v.red_insights_enabled); - } - - let settings: OrganizationSetting = match db::get(&key).await { - Ok(settings) => json::from_slice(&settings)?, - Err(Error::DbError(infra::errors::DbError::KeyNotExists(_))) => { - OrganizationSetting::default() - } - Err(e) => return Err(e), - }; - let red_insights_enabled = settings.red_insights_enabled; - - ORGANIZATION_SETTING - .write() - .await - .insert(key.to_string(), settings); - Ok(red_insights_enabled) -} - -/// Whether investigation threads show raw tool input and output; absent settings mean off. -pub async fn get_org_setting_ai_rca_thread_tool_output(org_id: &str) -> Result { - let key = format!("{ORG_SETTINGS_KEY_PREFIX}/{org_id}"); - if let Some(v) = ORGANIZATION_SETTING.read().await.get(&key) { - return Ok(v.ai_rca_thread_tool_output); - } - - let settings: OrganizationSetting = match db::get(&key).await { - Ok(settings) => json::from_slice(&settings)?, - Err(Error::DbError(infra::errors::DbError::KeyNotExists(_))) => { - OrganizationSetting::default() - } - Err(e) => return Err(e), - }; - let ai_rca_thread_tool_output = settings.ai_rca_thread_tool_output; - - ORGANIZATION_SETTING - .write() - .await - .insert(key.to_string(), settings); - Ok(ai_rca_thread_tool_output) -} - /// The coordinator delete evicts `ORGANIZATION_SETTING` on every node. pub async fn delete_org_setting(org_id: &str) -> errors::Result<()> { let key = format!("{ORG_SETTINGS_KEY_PREFIX}/{org_id}"); diff --git a/src/infra/src/table/cipher.rs b/src/infra/src/table/cipher.rs index 959c42cf3a..edddc67d0b 100644 --- a/src/infra/src/table/cipher.rs +++ b/src/infra/src/table/cipher.rs @@ -13,19 +13,13 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -use std::{ - sync::{LazyLock, OnceLock}, - time::{Duration, Instant}, -}; +use std::{sync::OnceLock, time::Duration}; use base64::{Engine, prelude::BASE64_STANDARD}; -use config::{ - RwHashMap, - utils::{ - encryption::{Algorithm, SECRET_PREFIX, decode_encryption_key}, - rand::random_bytes, - time::now_micros, - }, +use config::utils::{ + encryption::{Algorithm, SECRET_PREFIX, decode_encryption_key}, + rand::random_bytes, + time::now_micros, }; use sea_orm::{ ColumnTrait, ConnectionTrait, EntityTrait, Order, QueryFilter, QueryOrder, QuerySelect, Set, @@ -33,7 +27,7 @@ use sea_orm::{ }; use serde::{Deserialize, Serialize}; -use super::entity::cipher_keys::*; +use super::{entity::cipher_keys::*, ttl_cache::TtlCache}; use crate::{ db::{get_orm_client_ro, get_orm_client_rw}, errors, @@ -52,8 +46,8 @@ pub const CIPHER_KEY_PREFIX: &str = "/cipher_keys/"; /// public API. pub const DEFAULT_DEK_NAME: &str = "__default__"; -/// In-memory DEK cache: org → (raw key bytes, time of last load). -static DEK_CACHE: LazyLock, Instant)>> = LazyLock::new(Default::default); +/// In-memory DEK cache: org → raw key bytes. +static DEK_CACHE: TtlCache> = TtlCache::new(DEK_CACHE_TTL); const DEK_CACHE_TTL: Duration = Duration::from_secs(300); /// Key state, installed once by [`boot`] from `src/main.rs` and `tests/integration_test.rs`. @@ -406,14 +400,12 @@ pub async fn get_or_create_dek(org: &str) -> Result, errors::Error> { /// Returns the per-org DEK, using an in-memory cache with a 5-minute TTL to /// avoid a database round-trip on every encryption/decryption operation. pub async fn get_dek(org: &str) -> Result, errors::Error> { - if let Some(entry) = DEK_CACHE.get(org) - && entry.1.elapsed() < DEK_CACHE_TTL - { - return Ok(entry.0.clone()); + if let Some(cached) = DEK_CACHE.get(org) { + return Ok(cached); } let dek = get_or_create_dek(org).await?; - DEK_CACHE.insert(org.to_string(), (dek.clone(), Instant::now())); + DEK_CACHE.insert(org.to_string(), dek.clone()); Ok(dek) } @@ -488,10 +480,8 @@ pub async fn boot(key_b64: Option<&str>) -> Result<(), errors::Error> { } async fn find_dek(org: &str) -> Result>, errors::Error> { - if let Some(entry) = DEK_CACHE.get(org) - && entry.1.elapsed() < DEK_CACHE_TTL - { - return Ok(Some(entry.0.clone())); + if let Some(dek) = DEK_CACHE.get(org) { + return Ok(Some(dek)); } let Some(b64) = get_data(org, EntryKind::CipherKey, DEFAULT_DEK_NAME).await? else { return Ok(None); @@ -499,7 +489,7 @@ async fn find_dek(org: &str) -> Result>, errors::Error> { let dek = BASE64_STANDARD .decode(b64) .map_err(|e| errors::Error::Message(format!("failed to decode DEK for org {org}: {e}")))?; - DEK_CACHE.insert(org.to_string(), (dek.clone(), Instant::now())); + DEK_CACHE.insert(org.to_string(), dek.clone()); Ok(Some(dek)) } @@ -1014,7 +1004,7 @@ mod tests { ], ) .await; - DEK_CACHE.insert(org.to_string(), (vec![9u8], Instant::now())); + DEK_CACHE.insert(org.to_string(), vec![9u8]); let (first, installed) = boot_db(&db, aes(1)).await; assert_eq!(installed, Some(Ok("AES"))); @@ -1039,7 +1029,7 @@ mod tests { .iter() .all(|r| r.2.starts_with(SECRET_PREFIX)) ); - assert_eq!(DEK_CACHE.get(org).unwrap().0, vec![9u8]); + assert_eq!(DEK_CACHE.get(org).unwrap(), vec![9u8]); let before = stored(&db).await; let second = boot_db(&db, aes(1)).await.0.unwrap().unwrap(); diff --git a/src/infra/src/table/mod.rs b/src/infra/src/table/mod.rs index b2e709c589..6e4fc89fe4 100644 --- a/src/infra/src/table/mod.rs +++ b/src/infra/src/table/mod.rs @@ -133,6 +133,7 @@ pub mod timed_annotation_panels; pub mod timed_annotations; #[cfg(feature = "cloud")] pub mod trial_quota_usage; +mod ttl_cache; pub mod user_auth_state; pub mod user_password_history; pub mod users; diff --git a/src/infra/src/table/oncall_ownership.rs b/src/infra/src/table/oncall_ownership.rs index 39c3f7931a..92fb73a2fe 100644 --- a/src/infra/src/table/oncall_ownership.rs +++ b/src/infra/src/table/oncall_ownership.rs @@ -15,14 +15,10 @@ //! Ownership rules, and the queue of signals that fell through them — one question, twice. -use std::{ - collections::HashMap, - sync::LazyLock, - time::{Duration, Instant}, -}; +use std::{collections::HashMap, time::Duration}; use config::{ - RwHashMap, ider, + ider, meta::oncall::{OwnershipRule, SubjectType, UnroutedSignal, canonical_path}, utils::time::now_micros, }; @@ -31,7 +27,10 @@ use sea_orm::{ QuerySelect, Set, }; -use super::entity::{oncall_ownership_rules, oncall_unrouted_signals}; +use super::{ + entity::{oncall_ownership_rules, oncall_unrouted_signals}, + ttl_cache::TtlCache, +}; use crate::{db::get_orm_client_rw, errors}; /// An org's ownership rules, for the paging path (`06` §3). @@ -50,8 +49,7 @@ use crate::{db::get_orm_client_rw, errors}; /// /// Keyed by org, not by rule: the value is the whole set, which is what /// resolution needs. Backs [`list_cached`] only; the screens read [`list`]. -static RULES_CACHE: LazyLock, Instant)>> = - LazyLock::new(Default::default); +static RULES_CACHE: TtlCache> = TtlCache::new(RULES_CACHE_TTL); const RULES_CACHE_TTL: Duration = Duration::from_secs(60); @@ -187,13 +185,11 @@ pub async fn list(org_id: &str) -> Result, errors::Error> { /// anything here" is a stable answer and the common one for a young org, and it /// is exactly the read that would otherwise run on every firing. pub async fn list_cached(org_id: &str) -> Result, errors::Error> { - if let Some(entry) = RULES_CACHE.get(org_id) - && entry.1.elapsed() < RULES_CACHE_TTL - { - return Ok(entry.0.clone()); + if let Some(cached) = RULES_CACHE.get(org_id) { + return Ok(cached); } let rules = list(org_id).await?; - RULES_CACHE.insert(org_id.to_string(), (rules.clone(), Instant::now())); + RULES_CACHE.insert(org_id.to_string(), rules.clone()); Ok(rules) } diff --git a/src/infra/src/table/oncall_policies.rs b/src/infra/src/table/oncall_policies.rs index 3e61240076..4d2f903e5e 100644 --- a/src/infra/src/table/oncall_policies.rs +++ b/src/infra/src/table/oncall_policies.rs @@ -15,19 +15,16 @@ //! A team's escalation policy, stored as JSON. -use std::{ - sync::LazyLock, - time::{Duration, Instant}, -}; +use std::time::Duration; use config::{ - RwHashMap, ider, + ider, meta::oncall::{EscalationPolicy, PriorityRung}, utils::time::now_micros, }; use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, Set}; -use super::entity::oncall_policies; +use super::{entity::oncall_policies, ttl_cache::TtlCache}; use crate::{db::get_orm_client_rw, errors}; /// The team's escalation policy, for the paging path (`06` §3). @@ -43,8 +40,7 @@ use crate::{db::get_orm_client_rw, errors}; /// /// Backs [`get_or_create_cached`] only. The policy screen reads /// [`get_or_create`]. -static POLICY_CACHE: LazyLock> = - LazyLock::new(Default::default); +static POLICY_CACHE: TtlCache = TtlCache::new(POLICY_CACHE_TTL); const POLICY_CACHE_TTL: Duration = Duration::from_secs(60); @@ -61,7 +57,7 @@ pub fn invalidate_cache(org_id: &str, team_id: &str) { /// Drops every policy an org holds, for when the org itself is deleted. pub fn invalidate_org(org_id: &str) { let prefix = format!("{org_id}/"); - POLICY_CACHE.retain(|key, _| !key.starts_with(&prefix)); + POLICY_CACHE.retain(|key| !key.starts_with(&prefix)); } pub(super) async fn invalidate_and_publish(org_id: &str, team_id: &str) { @@ -176,13 +172,11 @@ pub async fn get_or_create_cached( team_id: &str, ) -> Result { let key = policy_cache_key(org_id, team_id); - if let Some(entry) = POLICY_CACHE.get(&key) - && entry.1.elapsed() < POLICY_CACHE_TTL - { - return Ok(entry.0.clone()); + if let Some(cached) = POLICY_CACHE.get(&key) { + return Ok(cached); } let policy = get_or_create(org_id, team_id).await?; - POLICY_CACHE.insert(key, (policy.clone(), Instant::now())); + POLICY_CACHE.insert(key, policy.clone()); Ok(policy) } @@ -428,14 +422,19 @@ mod tests { EscalationPolicy::default_for_team("pol_1", org, "team_1", "rot_primary", None) }; for org in ["inv_pol_a", "inv_pol_ab"] { - POLICY_CACHE.insert( - policy_cache_key(org, "team_1"), - (policy(org), Instant::now()), - ); + POLICY_CACHE.insert(policy_cache_key(org, "team_1"), policy(org)); } invalidate_org("inv_pol_a"); - assert!(!POLICY_CACHE.contains_key(&policy_cache_key("inv_pol_a", "team_1"))); - assert!(POLICY_CACHE.contains_key(&policy_cache_key("inv_pol_ab", "team_1"))); + assert!( + POLICY_CACHE + .get(&policy_cache_key("inv_pol_a", "team_1")) + .is_none() + ); + assert!( + POLICY_CACHE + .get(&policy_cache_key("inv_pol_ab", "team_1")) + .is_some() + ); invalidate_org("inv_pol_ab"); } } diff --git a/src/infra/src/table/oncall_schedules.rs b/src/infra/src/table/oncall_schedules.rs index 225c6006aa..d320d7e13c 100644 --- a/src/infra/src/table/oncall_schedules.rs +++ b/src/infra/src/table/oncall_schedules.rs @@ -15,19 +15,16 @@ //! A team's schedule: rotations in one JSON column, because they are read and written as a set. -use std::{ - sync::LazyLock, - time::{Duration, Instant}, -}; +use std::time::Duration; use config::{ - RwHashMap, ider, + ider, meta::oncall::{Rotation, Schedule, Unavailability}, utils::time::now_micros, }; use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, Set}; -use super::entity::oncall_schedules; +use super::{entity::oncall_schedules, ttl_cache::TtlCache}; use crate::{db::get_orm_client_rw, errors}; /// The team's schedule with its covers, for the paging path (`06` §3). @@ -48,8 +45,7 @@ use crate::{db::get_orm_client_rw, errors}; /// exists for, and `oncall_overrides` emits one on every write. /// /// Backs [`get_by_team_cached`] only. Every screen still reads [`get_by_team`]. -static SCHEDULE_CACHE: LazyLock> = - LazyLock::new(Default::default); +static SCHEDULE_CACHE: TtlCache = TtlCache::new(SCHEDULE_CACHE_TTL); /// One minute. See [`SCHEDULE_CACHE`] — the backstop for a coordinator event /// that never arrived, not the invalidation mechanism, and what it backstops is @@ -73,7 +69,7 @@ pub fn invalidate_cache(org_id: &str, team_id: &str) { /// single team to name. Bounded by the org's team count. pub fn invalidate_org(org_id: &str) { let prefix = format!("{org_id}/"); - SCHEDULE_CACHE.retain(|key, _| !key.starts_with(&prefix)); + SCHEDULE_CACHE.retain(|key| !key.starts_with(&prefix)); } /// Invalidates every schedule in the org locally, and tells every other node. @@ -257,14 +253,12 @@ pub async fn get_by_team_cached( team_id: &str, ) -> Result, errors::Error> { let key = schedule_cache_key(org_id, team_id); - if let Some(entry) = SCHEDULE_CACHE.get(&key) - && entry.1.elapsed() < SCHEDULE_CACHE_TTL - { - return Ok(Some(entry.0.clone())); + if let Some(cached) = SCHEDULE_CACHE.get(&key) { + return Ok(Some(cached)); } let found = get_by_team(org_id, team_id).await?; if let Some(schedule) = &found { - SCHEDULE_CACHE.insert(key, (schedule.clone(), Instant::now())); + SCHEDULE_CACHE.insert(key, schedule.clone()); } Ok(found) } diff --git a/src/infra/src/table/oncall_teams.rs b/src/infra/src/table/oncall_teams.rs index ceb103708e..1e73329fda 100644 --- a/src/infra/src/table/oncall_teams.rs +++ b/src/infra/src/table/oncall_teams.rs @@ -30,7 +30,10 @@ use sea_orm::{ TransactionTrait, }; -use super::entity::{oncall_team_members, oncall_teams}; +use super::{ + entity::{oncall_team_members, oncall_teams}, + ttl_cache::TtlCache, +}; use crate::{ db::{get_orm_client_ro, get_orm_client_rw}, errors, @@ -43,7 +46,7 @@ use crate::{ /// /// Backs [`get_cached`] only, never [`get`]: the screens that edit a team read /// through `get`, so an admin never sees what they just replaced. -static TEAM_CACHE: LazyLock> = LazyLock::new(Default::default); +static TEAM_CACHE: TtlCache = TtlCache::new(TEAM_CACHE_TTL); /// The roster, for the paging path. /// @@ -51,8 +54,7 @@ static TEAM_CACHE: LazyLock> = LazyLock::new( /// past the third rung. `06` §6 budgets five minutes of staleness because the /// consequence is bounded and one-directional: a member removed moments ago may /// receive one more page. Nothing here can suppress a page. -static MEMBERS_CACHE: LazyLock, Instant)>> = - LazyLock::new(Default::default); +static MEMBERS_CACHE: TtlCache> = TtlCache::new(TEAM_CACHE_TTL); /// Whether an org has any team at all, for the firing path. /// @@ -95,7 +97,7 @@ pub fn invalidate_members_cache(team_id: &str) { /// Drops every org-keyed on-call cache of a deleted org; rosters are keyed by team alone. pub fn invalidate_org(org_id: &str) { let prefix = format!("{org_id}/"); - TEAM_CACHE.retain(|key, _| !key.starts_with(&prefix)); + TEAM_CACHE.retain(|key| !key.starts_with(&prefix)); ORG_HAS_TEAMS_CACHE.remove(org_id); super::oncall_schedules::invalidate_org(org_id); super::oncall_policies::invalidate_org(org_id); @@ -376,14 +378,12 @@ pub async fn list_members(team_id: &str) -> Result, errors::Erro /// keep saying so. pub async fn get_cached(org_id: &str, id: &str) -> Result, errors::Error> { let key = team_cache_key(org_id, id); - if let Some(entry) = TEAM_CACHE.get(&key) - && entry.1.elapsed() < TEAM_CACHE_TTL - { - return Ok(Some(entry.0.clone())); + if let Some(cached) = TEAM_CACHE.get(&key) { + return Ok(Some(cached)); } let found = get(org_id, id).await?; if let Some(team) = &found { - TEAM_CACHE.insert(key, (team.clone(), Instant::now())); + TEAM_CACHE.insert(key, team.clone()); } Ok(found) } @@ -423,13 +423,11 @@ pub async fn has_any_cached(org_id: &str) -> Result { /// roster IS cached, unlike a missing team: "this team has nobody on it" is a /// real, stable answer, and it is what a coverage gap is made of. pub async fn list_members_cached(team_id: &str) -> Result, errors::Error> { - if let Some(entry) = MEMBERS_CACHE.get(team_id) - && entry.1.elapsed() < TEAM_CACHE_TTL - { - return Ok(entry.0.clone()); + if let Some(cached) = MEMBERS_CACHE.get(team_id) { + return Ok(cached); } let members = list_members(team_id).await?; - MEMBERS_CACHE.insert(team_id.to_string(), (members.clone(), Instant::now())); + MEMBERS_CACHE.insert(team_id.to_string(), members.clone()); Ok(members) } @@ -711,13 +709,21 @@ mod tests { }) }; for org in ["inv_team_a", "inv_team_ab"] { - TEAM_CACHE.insert(team_cache_key(org, "team_1"), (team(org), Instant::now())); + TEAM_CACHE.insert(team_cache_key(org, "team_1"), team(org)); ORG_HAS_TEAMS_CACHE.insert(org.to_string(), (true, Instant::now())); } invalidate_org("inv_team_a"); - assert!(!TEAM_CACHE.contains_key(&team_cache_key("inv_team_a", "team_1"))); + assert!( + TEAM_CACHE + .get(&team_cache_key("inv_team_a", "team_1")) + .is_none() + ); assert!(!ORG_HAS_TEAMS_CACHE.contains_key("inv_team_a")); - assert!(TEAM_CACHE.contains_key(&team_cache_key("inv_team_ab", "team_1"))); + assert!( + TEAM_CACHE + .get(&team_cache_key("inv_team_ab", "team_1")) + .is_some() + ); assert!(ORG_HAS_TEAMS_CACHE.contains_key("inv_team_ab")); invalidate_org("inv_team_ab"); } diff --git a/src/infra/src/table/org_ingestion_tokens.rs b/src/infra/src/table/org_ingestion_tokens.rs index 5a2841ae88..93588ef43a 100644 --- a/src/infra/src/table/org_ingestion_tokens.rs +++ b/src/infra/src/table/org_ingestion_tokens.rs @@ -13,19 +13,18 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -use std::{ - sync::LazyLock, - time::{Duration, Instant}, -}; +use std::time::Duration; -use config::RwHashMap; use sea_orm::{ ColumnTrait, EntityTrait, FromQueryResult, NotSet, Order, PaginatorTrait, QueryFilter, QueryOrder, QuerySelect, Set, entity::prelude::*, sea_query::OnConflict, }; use serde::{Deserialize, Deserializer, Serialize}; -use super::entity::org_ingestion_tokens::{ActiveModel, Column, Entity, Model}; +use super::{ + entity::org_ingestion_tokens::{ActiveModel, Column, Entity, Model}, + ttl_cache::TtlCache, +}; use crate::{ db::{get_orm_client_ro, get_orm_client_rw}, errors::{self, DbError, Error}, @@ -48,9 +47,8 @@ pub const ORG_INGESTION_TOKEN_PREFIX: &str = "o2oi_"; /// /// `None` is cached too, so an org with no enabled token does not re-query on /// every job. -static DEFAULT_TOKEN_CACHE: LazyLock< - RwHashMap, Instant)>, -> = LazyLock::new(Default::default); +static DEFAULT_TOKEN_CACHE: TtlCache> = + TtlCache::new(DEFAULT_TOKEN_CACHE_TTL); /// Backstop only. Cross-node invalidation arrives through the existing /// `/org_ingestion_tokens/` coordinator watch (see `db::org_ingestion_tokens::watch`), @@ -73,10 +71,8 @@ pub fn invalidate_default_cache(org_id: &str) { pub async fn find_default_enabled( org_id: &str, ) -> Result, errors::Error> { - if let Some(entry) = DEFAULT_TOKEN_CACHE.get(org_id) - && entry.1.elapsed() < DEFAULT_TOKEN_CACHE_TTL - { - return Ok(entry.0.clone()); + if let Some(cached) = DEFAULT_TOKEN_CACHE.get(org_id) { + return Ok(cached); } let client = get_orm_client_rw().await; @@ -90,7 +86,7 @@ pub async fn find_default_enabled( .map_err(|e| Error::DbError(DbError::SeaORMError(e.to_string())))? .map(OrgIngestionTokenRecord::from); - DEFAULT_TOKEN_CACHE.insert(org_id.to_string(), (record.clone(), Instant::now())); + DEFAULT_TOKEN_CACHE.insert(org_id.to_string(), record.clone()); Ok(record) } diff --git a/src/infra/src/table/synthetics_agents.rs b/src/infra/src/table/synthetics_agents.rs index 68d6746b65..fd0b5f6a01 100644 --- a/src/infra/src/table/synthetics_agents.rs +++ b/src/infra/src/table/synthetics_agents.rs @@ -19,17 +19,16 @@ //! `/synthetics/agent/register`; `last_seen_at` refreshed by register and by //! every job lease. A location whose agents are all stale is reported "down". -use std::{ - sync::LazyLock, - time::{Duration, Instant}, -}; +use std::time::Duration; -use config::RwHashMap; use sea_orm::{ ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect, Set, sea_query::Expr, }; -use super::entity::synthetics_agents::{ActiveModel, Column, Entity, Model}; +use super::{ + entity::synthetics_agents::{ActiveModel, Column, Entity, Model}, + ttl_cache::TtlCache, +}; use crate::{ db::{get_orm_client_ro, get_orm_client_rw}, errors::{self, DbError, Error}, @@ -78,8 +77,7 @@ impl From for SyntheticsAgentRecord { /// not invalidate — invalidating there would defeat the cache entirely, since it /// fires at exactly the rate of the reads being served. Anything that judges /// agent liveness (the staleness watcher, location health) must use [`get`]. -static AGENT_CACHE: LazyLock> = - LazyLock::new(Default::default); +static AGENT_CACHE: TtlCache = TtlCache::new(AGENT_CACHE_TTL); const AGENT_CACHE_TTL: Duration = Duration::from_secs(15); @@ -255,14 +253,12 @@ pub async fn count_by_token( /// Reads an agent from cache when fresh — see [`AGENT_CACHE`] for the staleness /// contract. Use [`get`] when `last_seen_at` matters. pub async fn get_cached(agent_id: &str) -> Result, errors::Error> { - if let Some(entry) = AGENT_CACHE.get(agent_id) - && entry.1.elapsed() < AGENT_CACHE_TTL - { - return Ok(Some(entry.0.clone())); + if let Some(cached) = AGENT_CACHE.get(agent_id) { + return Ok(Some(cached)); } let found = get(agent_id).await?; if let Some(rec) = &found { - AGENT_CACHE.insert(agent_id.to_string(), (rec.clone(), Instant::now())); + AGENT_CACHE.insert(agent_id.to_string(), rec.clone()); } Ok(found) } diff --git a/src/infra/src/table/synthetics_checks.rs b/src/infra/src/table/synthetics_checks.rs index d19c4b4535..29ea7415e0 100644 --- a/src/infra/src/table/synthetics_checks.rs +++ b/src/infra/src/table/synthetics_checks.rs @@ -13,17 +13,11 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -use std::{ - sync::LazyLock, - time::{Duration, Instant}, -}; +use std::time::Duration; -use config::{ - RwHashMap, - meta::synthetics::{ - BrowserConfig, ListSyntheticsParams, Synthetic, SyntheticAuth, SyntheticCookie, - SyntheticFrequency, SyntheticSettings, SyntheticStatus, SyntheticType, SyntheticVariable, - }, +use config::meta::synthetics::{ + BrowserConfig, ListSyntheticsParams, Synthetic, SyntheticAuth, SyntheticCookie, + SyntheticFrequency, SyntheticSettings, SyntheticStatus, SyntheticType, SyntheticVariable, }; use sea_orm::{ ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, PaginatorTrait, QueryFilter, @@ -31,7 +25,10 @@ use sea_orm::{ }; use serde::{Deserialize, Serialize}; -use super::entity::synthetics_checks::{self, ActiveModel, Column, Entity}; +use super::{ + entity::synthetics_checks::{self, ActiveModel, Column, Entity}, + ttl_cache::TtlCache, +}; use crate::errors; /// `org_id/synthetics_id` → the synthetic check **definition**, with load time. @@ -48,8 +45,7 @@ use crate::errors; /// make the cache useless (they fire at the same rate as the reads it serves), /// so instead they deliberately do **not** invalidate — and callers that need /// scheduling or status state must use [`get`], `fetch_due` or `get_alert_state`. -static SYNTHETIC_CACHE: LazyLock> = - LazyLock::new(Default::default); +static SYNTHETIC_CACHE: TtlCache = TtlCache::new(SYNTHETIC_CACHE_TTL); const SYNTHETIC_CACHE_TTL: Duration = Duration::from_secs(15); @@ -207,15 +203,13 @@ pub async fn get_cached( id: &str, ) -> Result, errors::Error> { let key = synthetic_cache_key(org_id, id); - if let Some(entry) = SYNTHETIC_CACHE.get(&key) - && entry.1.elapsed() < SYNTHETIC_CACHE_TTL - { - return Ok(Some(entry.0.clone())); + if let Some(cached) = SYNTHETIC_CACHE.get(&key) { + return Ok(Some(cached)); } let found = get(conn, org_id, id).await?; if let Some(synthetic) = &found { - SYNTHETIC_CACHE.insert(key, (synthetic.clone(), Instant::now())); + SYNTHETIC_CACHE.insert(key, synthetic.clone()); } Ok(found) } diff --git a/src/infra/src/table/synthetics_probe_tokens.rs b/src/infra/src/table/synthetics_probe_tokens.rs index 4c69442820..70e5a75758 100644 --- a/src/infra/src/table/synthetics_probe_tokens.rs +++ b/src/infra/src/table/synthetics_probe_tokens.rs @@ -19,17 +19,16 @@ //! (`/synthetics/jobs/resolve`, `/ack`, `/lease`). Separate from //! `org_ingestion_tokens` (`o2oi_`) which is write-only ingest. -use std::{ - sync::LazyLock, - time::{Duration, Instant}, -}; +use std::time::Duration; -use config::RwHashMap; use sea_orm::{ ColumnTrait, EntityTrait, QueryFilter, Set, SqlErr, TransactionTrait, sea_query::Expr, }; -use super::entity::synthetics_probe_tokens::{ActiveModel, Column, Entity, Model}; +use super::{ + entity::synthetics_probe_tokens::{ActiveModel, Column, Entity, Model}, + ttl_cache::TtlCache, +}; use crate::{ db::{get_orm_client_ro, get_orm_client_rw}, errors::{self, DbError, Error}, @@ -53,13 +52,11 @@ pub const SYNTHETICS_PROBE_TOKEN_PREFIX: &str = "o2syn_"; /// effectively immediate for an operator and long enough to remove ~99 % of the /// queries. Writes in this module invalidate eagerly, so the TTL only covers a /// disable performed on a *different* node. -static TOKEN_CACHE: LazyLock, Instant)>> = - LazyLock::new(Default::default); +static TOKEN_CACHE: TtlCache> = TtlCache::new(TOKEN_CACHE_TTL); /// Org → its default enabled token, same TTL rules as [`TOKEN_CACHE`]. -static DEFAULT_TOKEN_CACHE: LazyLock< - RwHashMap, Instant)>, -> = LazyLock::new(Default::default); +static DEFAULT_TOKEN_CACHE: TtlCache> = + TtlCache::new(TOKEN_CACHE_TTL); const TOKEN_CACHE_TTL: Duration = Duration::from_secs(10); @@ -167,10 +164,8 @@ pub async fn add(record: &SyntheticsProbeTokenRecord) -> Result<(), errors::Erro /// path (the tenant boundary). Matches ANY enabled token, so old + new tokens /// coexist during a rotation overlap window. pub async fn find_global(token: &str) -> Result, errors::Error> { - if let Some(entry) = TOKEN_CACHE.get(token) - && entry.1.elapsed() < TOKEN_CACHE_TTL - { - return Ok(entry.0.clone()); + if let Some(cached) = TOKEN_CACHE.get(token) { + return Ok(cached); } let client = get_orm_client_rw().await; @@ -182,7 +177,7 @@ pub async fn find_global(token: &str) -> Result Result Result, errors::Error> { - if let Some(entry) = DEFAULT_TOKEN_CACHE.get(org_id) - && entry.1.elapsed() < TOKEN_CACHE_TTL - { - return Ok(entry.0.clone()); + if let Some(cached) = DEFAULT_TOKEN_CACHE.get(org_id) { + return Ok(cached); } let client = get_orm_client_rw().await; @@ -207,7 +200,7 @@ pub async fn find_default( .map_err(|e| Error::DbError(DbError::SeaORMError(e.to_string())))? .map(SyntheticsProbeTokenRecord::from); - DEFAULT_TOKEN_CACHE.insert(org_id.to_string(), (record.clone(), Instant::now())); + DEFAULT_TOKEN_CACHE.insert(org_id.to_string(), record.clone()); Ok(record) } diff --git a/src/infra/src/table/synthetics_variables.rs b/src/infra/src/table/synthetics_variables.rs index 204089b8f8..d773c2a4ac 100644 --- a/src/infra/src/table/synthetics_variables.rs +++ b/src/infra/src/table/synthetics_variables.rs @@ -13,18 +13,15 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -use std::{ - sync::LazyLock, - time::{Duration, Instant}, -}; +use std::time::Duration; -use config::{ - RwHashMap, - meta::synthetics_variables::{SyntheticsVariableView, VariableValueView}, -}; +use config::meta::synthetics_variables::{SyntheticsVariableView, VariableValueView}; use sea_orm::{ColumnTrait, ConnectionTrait, EntityTrait, QueryFilter, QueryOrder, Set, SqlErr}; -use super::entity::synthetics_variables::{ActiveModel, Column, Entity, Model}; +use super::{ + entity::synthetics_variables::{ActiveModel, Column, Entity, Model}, + ttl_cache::TtlCache, +}; use crate::errors::{self, DbError, Error}; /// The stored `kind` for a write-only variable. @@ -33,8 +30,7 @@ pub const KIND_SECRET: &str = "secret"; pub const KIND_PLAIN: &str = "plain"; /// `org_id` → every shared variable in the org, with the time it was loaded. -static VARIABLE_CACHE: LazyLock, Instant)>> = - LazyLock::new(Default::default); +static VARIABLE_CACHE: TtlCache> = TtlCache::new(VARIABLE_CACHE_TTL); const VARIABLE_CACHE_TTL: Duration = Duration::from_secs(15); @@ -131,13 +127,11 @@ pub async fn list_cached( conn: &C, org_id: &str, ) -> Result, errors::Error> { - if let Some(entry) = VARIABLE_CACHE.get(org_id) - && entry.1.elapsed() < VARIABLE_CACHE_TTL - { - return Ok(entry.0.clone()); + if let Some(cached) = VARIABLE_CACHE.get(org_id) { + return Ok(cached); } let rows = list(conn, org_id).await?; - VARIABLE_CACHE.insert(org_id.to_string(), (rows.clone(), Instant::now())); + VARIABLE_CACHE.insert(org_id.to_string(), rows.clone()); Ok(rows) } diff --git a/src/infra/src/table/ttl_cache.rs b/src/infra/src/table/ttl_cache.rs new file mode 100644 index 0000000000..8dad9e01b3 --- /dev/null +++ b/src/infra/src/table/ttl_cache.rs @@ -0,0 +1,93 @@ +// Copyright 2026 OpenObserve Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +use std::{ + sync::LazyLock, + time::{Duration, Instant}, +}; + +use config::RwHashMap; + +/// Entries read as missing past the TTL; stale ones are never swept, only overwritten. +pub(crate) struct TtlCache { + ttl: Duration, + entries: LazyLock>, +} + +impl TtlCache { + pub(crate) const fn new(ttl: Duration) -> Self { + Self { + ttl, + entries: LazyLock::new(RwHashMap::default), + } + } + + pub(crate) fn get(&self, key: &str) -> Option { + self.entries + .get(key) + .filter(|entry| entry.1.elapsed() < self.ttl) + .map(|entry| entry.0.clone()) + } + + pub(crate) fn insert(&self, key: String, value: V) { + self.entries.insert(key, (value, Instant::now())); + } + + pub(crate) fn remove(&self, key: &str) { + self.entries.remove(key); + } + + pub(crate) fn retain(&self, mut keep: impl FnMut(&str) -> bool) { + self.entries.retain(|key, _| keep(key)); + } + + pub(crate) fn clear(&self) { + self.entries.clear(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_fresh_entry_is_returned() { + let cache = TtlCache::new(Duration::from_secs(60)); + cache.insert("k".to_string(), 7); + assert_eq!(cache.get("k"), Some(7)); + } + + #[test] + fn an_entry_past_the_ttl_reads_as_missing() { + let cache = TtlCache::new(Duration::ZERO); + cache.insert("k".to_string(), 7); + assert_eq!(cache.get("k"), None); + } + + #[test] + fn remove_retain_and_clear_drop_entries() { + let cache = TtlCache::new(Duration::from_secs(60)); + for key in ["org/a", "org/b", "other/c", "x"] { + cache.insert(key.to_string(), 1); + } + cache.remove("x"); + assert_eq!(cache.get("x"), None); + cache.retain(|key| !key.starts_with("org/")); + assert_eq!(cache.get("org/a"), None); + assert_eq!(cache.get("other/c"), Some(1)); + cache.clear(); + assert_eq!(cache.get("other/c"), None); + } +} diff --git a/src/jobs/src/job/red_insights.rs b/src/jobs/src/job/red_insights.rs index 4c5dd08b9f..81c78f0408 100644 --- a/src/jobs/src/job/red_insights.rs +++ b/src/jobs/src/job/red_insights.rs @@ -63,15 +63,16 @@ pub async fn run() { } async fn reconcile_if_needed(org_id: &str) -> anyhow::Result<()> { - let prefetched = if db::organization::get_org_setting_red_insights_enabled(org_id).await? { - None - } else { - let configs = managed_configs(org_id).await?; - if configs.is_empty() { - return Ok(()); - } - Some(configs) - }; + let prefetched = + if db::organization::read_org_setting(org_id, |s| s.red_insights_enabled).await? { + None + } else { + let configs = managed_configs(org_id).await?; + if configs.is_empty() { + return Ok(()); + } + Some(configs) + }; let locker = infra::dist_lock::lock(&format!("red_insights/{org_id}"), 0).await?; let result = reconcile(org_id, prefetched).await; if let Err(e) = infra::dist_lock::unlock(&locker).await { @@ -85,7 +86,7 @@ async fn reconcile(org_id: &str, prefetched: Option>) -> Some(configs) => configs, None => managed_configs(org_id).await?, }; - if !db::organization::get_org_setting_red_insights_enabled(org_id).await? { + if !db::organization::read_org_setting(org_id, |s| s.red_insights_enabled).await? { for config in &configs { delete(org_id, config_str(config, "anomaly_id")).await; }