diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 493abd172e..dc3b9b6e8b 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -4380,6 +4380,23 @@ pub async fn monitor_db( } }; + // Delete the AI session backups older than their workspace's retention. Every ~40 min + // (240 iterations at the default 10 s, the most a u8 `should_run` counts): the retention + // counts in days. Spawned for the same reason as the credential maintenance above, a + // sweep of many sessions outlasting the join's deadline; the sweep's own advisory lock + // keeps one server at a time at it. + let ai_session_retention_f = async { + #[cfg(feature = "parquet")] + if server_mode && iteration.is_some() && iteration.as_ref().unwrap().should_run(240) { + if let Some(db) = conn.as_sql() { + let db = db.clone(); + tokio::spawn( + async move { windmill_api::sweep_expired_ai_session_backups(&db).await }, + ); + } + } + }; + // run every 2 iterations (~20s at the default LISTEN_NEW_EVENTS_INTERVAL_SEC). // Enterprise feature: the active `// freshness` backstop lives in // windmill-queue's `freshness_watchdog` (`private`); OSS gets a no-op stub. @@ -4434,6 +4451,7 @@ pub async fn monitor_db( cleanup_scheduled_job_deletions_f, git_auto_pull_f, git_credential_maintenance_f, + ai_session_retention_f, pipeline_freshness_watchdog_f, reconcile_unarmed_schedules_f, ); diff --git a/backend/tests/ai_sessions.rs b/backend/tests/ai_sessions.rs index 3eb28cfae8..6b33a4141e 100644 --- a/backend/tests/ai_sessions.rs +++ b/backend/tests/ai_sessions.rs @@ -1235,6 +1235,161 @@ async fn test_backup_writes_are_refused_for_the_wrong_owner_token_or_id( Ok(()) } +/// Sets the object's modification time `days` back: the FilesystemStorage answers +/// `last_modified` from it, so this is a session no push touched since. +fn age_object(path: &std::path::Path, days: u64) -> std::io::Result<()> { + let at = std::time::SystemTime::now() - std::time::Duration::from_secs(days * 86_400); + std::fs::File::options() + .write(true) + .open(path)? + .set_modified(at) +} + +#[sqlx::test(fixtures("base"))] +async fn test_expired_backups_are_swept_by_age_and_left_out_of_the_listing( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let base = format!( + "http://localhost:{}/api/w/test-workspace", + server.addr.port() + ); + let storage_dir = tempfile::tempdir()?; + configure_primary_lfs(&db, &storage_dir.path().to_string_lossy()).await?; + + let chat = |sid: &str, cid: &str| { + json!({ "id": cid, "record": { "id": cid, "sessionId": sid, "lastModified": 2, + "actualMessages": [], "displayMessages": [] } }) + }; + let whole = |sid: &str| { + json!({ + "id": sid, "whole": true, "epoch": 0, + "head": { "id": sid, "workspace_id": "test-workspace", "createdAt": 1, "chatId": "c1" }, + "chats": [chat(sid, "c1")], + "images": [{ "chat_id": "c1", "id": "img1", "data_url": "data:image/png;base64,AAAA" }], + "artifacts": { "items": [], "versions": [] } + }) + }; + // Two pushes split over parts of which only the first part landed: one a browser + // abandoned long ago (its token aged past the retention), one still in flight. + let opening = |sid: &str| { + json!({ + "id": sid, "whole": true, "epoch": 0, "push": format!("t-{sid}"), "opens": true, + "partial": true, "chats": [chat(sid, "c1")], + "head": { "id": sid, "workspace_id": "test-workspace", "createdAt": 1, "chatId": "c1" } + }) + }; + let resp = push( + &base, + "SECRET_TOKEN", + json!({ "owner": "test@windmill.dev", "sessions": [ + whole("old"), whole("live"), opening("abandoned"), opening("inflight") + ] }), + ) + .await?; + assert_eq!(resp.status(), 200, "{}", resp.text().await?); + let root = user_root(storage_dir.path(), "test@windmill.dev"); + age_object(&root.join("index/old/0"), 40)?; + age_object(&root.join("index/abandoned/push"), 40)?; + + let listed = |listing: Value| -> Vec { + let mut ids: Vec = listing["sessions"] + .as_array() + .unwrap() + .iter() + .map(|s| s["id"].as_str().unwrap().to_string()) + .collect(); + ids.sort(); + ids + }; + let objects = |root: &std::path::Path| -> Vec { + files_under(root) + .into_iter() + .map(|(p, _)| p.strip_prefix(root).unwrap().to_string_lossy().into_owned()) + .collect() + }; + + // Without a retention nothing is swept, however old. + windmill_api::sweep_expired_ai_session_backups(&db).await; + assert_eq!(listed(list(&base, "SECRET_TOKEN").await?), ["live", "old"]); + + let set_retention = |days: Value| { + authed( + client().post(format!("{base}/workspaces/edit_copilot_config")), + "SECRET_TOKEN", + ) + .json(&json!({ "sessions_retention_days": days })) + .send() + }; + let resp = set_retention(json!(0)).await?; + assert_eq!(resp.status(), 400, "{}", resp.text().await?); + let resp = set_retention(json!(30)).await?; + assert_eq!(resp.status(), 200, "{}", resp.text().await?); + + // The listing leaves the expired session out before the sweep reaches it. + assert_eq!(listed(list(&base, "SECRET_TOKEN").await?), ["live"]); + assert!(root.join("sessions/old/head.json").exists()); + + // A removal cut short (a directory stands where the head is, so it cannot be unlinked) + // leaves the sweep's record with the markers gone; the next pass finds it and finishes. + let head = root.join("sessions/old/head.json"); + std::fs::remove_file(&head)?; + std::fs::create_dir(&head)?; + std::fs::write(head.join("planted"), b"")?; + windmill_api::sweep_expired_ai_session_backups(&db).await; + assert!(root.join("index/old/sweep").exists()); + assert!(!root.join("index/old/0").exists()); + assert!(root.join("sessions/old/chats/c1.json").exists()); + assert_eq!(listed(list(&base, "SECRET_TOKEN").await?), ["live"]); + std::fs::remove_dir_all(&head)?; + + windmill_api::sweep_expired_ai_session_backups(&db).await; + let remaining = objects(&root); + assert!( + remaining + .iter() + .all(|p| !p.contains("/old/") && !p.contains("/abandoned/")), + "{remaining:?}" + ); + for kept in [ + "index/live/0", + "sessions/live/head.json", + "sessions/live/chats/c1.json", + "images/live/c1/img1", + "index/inflight/push", + "sessions/inflight/head.json", + ] { + assert!( + remaining.iter().any(|p| p == kept), + "{kept} in {remaining:?}" + ); + } + assert_eq!(listed(list(&base, "SECRET_TOKEN").await?), ["live"]); + + // A second pass has nothing to do; a session pushed again since its marker aged is + // renewed by the push, which rewrites the marker. + windmill_api::sweep_expired_ai_session_backups(&db).await; + age_object(&root.join("index/live/0"), 40)?; + let resp = push( + &base, + "SECRET_TOKEN", + json!({ "owner": "test@windmill.dev", + "sessions": [{ "id": "live", "epoch": 0, "chats": [chat("live", "c2")] }] }), + ) + .await?; + assert_eq!(resp.status(), 200, "{}", resp.text().await?); + windmill_api::sweep_expired_ai_session_backups(&db).await; + let mut after = objects(&root); + after.sort(); + let mut expected = remaining.clone(); + expected.push("sessions/live/chats/c2.json".to_string()); + expected.sort(); + assert_eq!(after, expected); + assert_eq!(listed(list(&base, "SECRET_TOKEN").await?), ["live"]); + Ok(()) +} + /// Puts the process-wide instance store back to none, even when an assertion fails. struct ResetInstanceStore; impl Drop for ResetInstanceStore { @@ -1333,6 +1488,23 @@ async fn test_backups_fall_back_to_the_instance_storage(db: Pool) -> a .unwrap_or_else(|| panic!("no fallback usage in {usage}")); assert!(fallback_usage["bytes"].as_i64().unwrap() > 0); + // The retention sweep reaches what the instance store keeps for the workspace, choosing + // that store from the row it reads the generation from. + let instance_user = user_root(instance_dir.path(), "test@windmill.dev"); + age_object(&instance_user.join("index/s1/0"), 40)?; + sqlx::query( + "UPDATE workspace_settings SET ai_config = '{\"sessions_retention_days\": 30}' \ + WHERE workspace_id = 'test-workspace'", + ) + .execute(&db) + .await?; + windmill_api::sweep_expired_ai_session_backups(&db).await; + assert!(!instance_user.join("index/s1/0").exists()); + assert!(!instance_user.join("sessions/s1/head.json").exists()); + // Pushed again, so the rotation below has a backup to delete. + assert_eq!(push_whole().await?.status(), 200); + assert!(!files_under(&in_instance).is_empty()); + // A key rotation sweeps the older generation out of the instance store too. rotate(&base, &"c".repeat(64)).await?; wait_until_empty(&in_instance, "a rotation on the instance store").await; diff --git a/backend/windmill-api-workspaces/src/ai_session_backups.rs b/backend/windmill-api-workspaces/src/ai_session_backups.rs index 46c91d5354..8e7b75e1e2 100644 --- a/backend/windmill-api-workspaces/src/ai_session_backups.rs +++ b/backend/windmill-api-workspaces/src/ai_session_backups.rs @@ -123,8 +123,9 @@ pub async fn fallback_store(db: &DB) -> Result> { } /// The workspace's primary storage, resolved without a caller: a rotation runs its deletion -/// off its own request. -async fn primary_store(db: &DB, w_id: &str) -> Result> { +/// off its own request, and the retention sweep off any. The caller must be the server +/// itself; nothing here checks who asks. +pub async fn primary_store(db: &DB, w_id: &str) -> Result> { let Some(lfs_json) = sqlx::query_scalar!( "SELECT large_file_storage FROM workspace_settings WHERE workspace_id = $1", w_id diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index c240948105..c29d99d05b 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -5629,6 +5629,14 @@ struct SessionWorkspaceStatusRequest { workspace_ids: Vec, } +/// `ai_config.sessions_retention_days` as stored, `None` when unset or not a count of days. +pub fn sessions_retention_days(value: Option<&serde_json::Value>) -> Option { + value + .and_then(|v| v.as_u64()) + .filter(|days| *days >= 1) + .and_then(|days| u32::try_from(days).ok()) +} + /// Reconciliation support for client-side AI sessions, which the backend cannot touch /// directly. The client posts the workspace ids its sessions reference and uses the /// per-id status to keep sessions in sync with workspace lifecycle: `deleted` (no row, or diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 63e8dd46ea..9ce1aef272 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -28448,6 +28448,14 @@ components: description: >- Stops browsers from backing their AI sessions up to the workspace's object storage. Read from the workspace's own settings like `copilot_disabled`. + sessions_retention_days: + type: integer + minimum: 1 + maximum: 3650 + description: >- + The server deletes the backup of a session no push has reached for this many + days. Unset keeps backups until the user deletes the session. Read from the + workspace's own settings like `copilot_disabled`. AISessionBackupListing: type: object diff --git a/backend/windmill-api/src/ai.rs b/backend/windmill-api/src/ai.rs index 8b9c7929ab..1710db9cec 100644 --- a/backend/windmill-api/src/ai.rs +++ b/backend/windmill-api/src/ai.rs @@ -455,6 +455,12 @@ pub struct AIConfig { /// (`ai_sessions.rs`). Read from the workspace's own row like `copilot_disabled`. #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub sessions_storage_disabled: bool, + /// The server's sweep (`ai_sessions.rs`) deletes the backup of a session no push has + /// reached for this many days. The copies in members' browsers are untouched. Unset + /// keeps backups until the user deletes the session. Read from the workspace's own row + /// like `copilot_disabled`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub sessions_retention_days: Option, } /// Negotiated rates in USD per million tokens. An unset cache rate is read as the @@ -491,6 +497,9 @@ impl ModelPriceOverride { } } +/// Ten years: past any plausible retention, and well within what a day count is turned into. +pub const MAX_SESSIONS_RETENTION_DAYS: u32 = 3650; + impl AIConfig { pub fn validate_model_pricing(&self) -> Result<()> { for (key, price) in self.model_pricing.iter().flatten() { @@ -499,6 +508,17 @@ impl AIConfig { Ok(()) } + pub fn validate_sessions_retention(&self) -> Result<()> { + match self.sessions_retention_days { + Some(days) if !(1..=MAX_SESSIONS_RETENTION_DAYS).contains(&days) => { + Err(Error::BadRequest(format!( + "AI session retention must be between 1 and {MAX_SESSIONS_RETENTION_DAYS} days (got {days})" + ))) + } + _ => Ok(()), + } + } + pub fn has_providers(&self) -> bool { self.providers .as_ref() diff --git a/backend/windmill-api/src/ai_sessions.rs b/backend/windmill-api/src/ai_sessions.rs index 776b3a0561..65a8137c4e 100644 --- a/backend/windmill-api/src/ai_sessions.rs +++ b/backend/windmill-api/src/ai_sessions.rs @@ -30,8 +30,9 @@ use serde_json::value::RawValue; use std::sync::Arc; use windmill_api_auth::is_effectively_unscoped; use windmill_api_workspaces::ai_session_backups::{ - fallback_store, generation_prefix, storage_id, MAX_OBJECT_BYTES, + fallback_store, generation_prefix, primary_store, storage_id, MAX_OBJECT_BYTES, }; +use windmill_api_workspaces::workspaces::sessions_retention_days; use windmill_common::error::{Error, JsonResult, Result}; use windmill_common::utils::calculate_hash; use windmill_common::variables::{crypt_from_key_with_suffix, get_workspace_key}; @@ -65,6 +66,46 @@ const MAX_LISTED_OBJECTS: usize = 5000; const MAX_LIST_SCAN: usize = 50_000; const LIST_MAX: usize = 500; const IO_CONCURRENCY: usize = 8; +/// Sessions the retention sweep deletes per workspace and pass at most; the rest wait for +/// the next pass. +const SWEEP_MAX_PER_WORKSPACE: usize = 1000; +/// Session-level advisory lock of the retention sweep: one server at a time runs it. +const SWEEP_LOCK_ID: i64 = 0x5745_4550_4149; +/// The name of the sweep's record next to a session's markers (see `Backend::sweep_key`). +const SWEEP_RECORD: &str = "sweep"; +/// The name of a split push's token next to a session's markers (see `Backend::push_key`). +const PUSH_TOKEN: &str = "push"; + +/// A marker modified before this is past a retention of `days`. +fn retention_cutoff(days: u32) -> chrono::DateTime { + chrono::Utc::now() - chrono::Duration::days(i64::from(days)) +} + +/// What a key under the `index/` prefix is. +enum IndexEntry { + /// The marker that lists the session, named by its epoch. + Marker(u32), + /// The retention sweep's record (see `Backend::sweep_key`). + Sweep, + /// The token of a push split over parts (see `Backend::push_key`). + Push, +} + +/// The session a key under the `index/` prefix belongs to, and what the key is. +fn index_entry<'a>(index: &ObjectPath, key: &'a ObjectPath) -> Option<(&'a str, IndexEntry)> { + // `Path` drops the trailing delimiter, so the remainder starts with one. + let rel = key.as_ref().strip_prefix(index.as_ref())?; + let (sid, name) = rel.trim_start_matches('/').split_once('/')?; + if sid.is_empty() { + return None; + } + let entry = match name { + SWEEP_RECORD => IndexEntry::Sweep, + PUSH_TOKEN => IndexEntry::Push, + epoch => IndexEntry::Marker(epoch.parse().ok()?), + }; + Some((sid, entry)) +} pub fn workspaced_service() -> Router { Router::new() @@ -101,6 +142,9 @@ struct Backend { /// owed to the storage alone (a rotation deleted the older generation's copy anyway). storage_id: String, generation: i64, + /// `ai_config.sessions_retention_days`: a session whose marker is older is not listed, + /// whether or not the sweep has deleted it yet. + retention_days: Option, /// The store is the instance object store standing in for a workspace without storage /// of its own (`ai_session_backups::fallback_store`). fallback: bool, @@ -111,6 +155,12 @@ impl Backend { ObjectPath::from(format!("{}/index/", self.prefix)) } + /// The moment a marker's modification time must reach to count as live, under the + /// workspace's retention; `None` without one. + fn retention_cutoff(&self) -> Option> { + self.retention_days.map(retention_cutoff) + } + /// The marker that lists the session, named by the session's move count so that of a /// session two workspaces list, the copy moved last is told from the listing alone. fn index_key(&self, sid: &str, epoch: u32) -> ObjectPath { @@ -121,10 +171,18 @@ impl Backend { ObjectPath::from(format!("{}/index/{sid}/", self.prefix)) } - /// The token of the push split over parts in progress, under the session so a removal - /// or the next whole push clears it with the rest. + /// Written by the retention sweep before it deletes anything of a session, and deleted + /// last (`remove_session`): what finds a removal the sweep started and could not finish, + /// the markers being gone by then. Not an epoch, so nothing lists or pulls a session by it. + fn sweep_key(&self, sid: &str) -> ObjectPath { + ObjectPath::from(format!("{}/index/{sid}/{SWEEP_RECORD}", self.prefix)) + } + + /// The token of the push split over parts in progress, next to the markers so a removal + /// or the next whole push clears it with them, and the retention sweep, which walks the + /// markers, finds one a browser abandoned. fn push_key(&self, sid: &str) -> ObjectPath { - ObjectPath::from(format!("{}/sessions/{sid}/push", self.prefix)) + ObjectPath::from(format!("{}/index/{sid}/{PUSH_TOKEN}", self.prefix)) } fn session_prefix(&self, sid: &str) -> ObjectPath { @@ -310,7 +368,12 @@ impl Backend { let mut stream = self.store.list(Some(&prefix)); while let Some(meta) = stream.next().await { let meta = meta.map_err(object_store_error_to_error)?; - listed |= marker; + // The sweep's record is not a marker: a session it started removing is absent. + listed |= marker + && meta + .location + .filename() + .is_some_and(|name| name.parse::().is_ok()); acc = fold( acc, meta.location.as_ref(), @@ -415,17 +478,21 @@ fn require_json_object(kind: &str, raw: &RawValue, max_bytes: usize) -> Result<( /// no instance store to stand in, or the admin switched them off. Both read as /// `enabled: false` so the browser stops trying. async fn backend(authed: &ApiAuthed, db: &DB, w_id: &str) -> Result> { - let (disabled, generation, has_storage) = sqlx::query_as::<_, (Option, i64, bool)>( - "SELECT (ai_config->>'sessions_storage_disabled')::bool, ai_sessions_backup_generation, \ - large_file_storage IS NOT NULL FROM workspace_settings WHERE workspace_id = $1", - ) - .bind(w_id) - .fetch_optional(db) - .await? - .unwrap_or((None, 0, false)); + let (disabled, retention, generation, has_storage) = + sqlx::query_as::<_, (Option, Option, i64, bool)>( + "SELECT (ai_config->>'sessions_storage_disabled')::bool, \ + ai_config->'sessions_retention_days', ai_sessions_backup_generation, \ + large_file_storage IS NOT NULL \ + FROM workspace_settings WHERE workspace_id = $1", + ) + .bind(w_id) + .fetch_optional(db) + .await? + .unwrap_or((None, None, 0, false)); if disabled.unwrap_or(false) { return Ok(None); } + let retention_days = sessions_retention_days(retention.as_ref()); // Decided from the row the generation came from: the instance store is written only // under a generation read while the workspace had no storage of its own, which // configuring one moves past (`ai_session_backups`). @@ -459,6 +526,7 @@ async fn backend(authed: &ApiAuthed, db: &DB, w_id: &str) -> Result, @@ -511,6 +581,7 @@ async fn list( })); }; let prefix = backend.index_prefix(); + let cutoff = backend.retention_cutoff(); let mut stream = backend.store.list(Some(&prefix)); // One marker per session, whatever the session holds: the newest LIST_MAX are kept as // the scan goes (a min-heap drops the oldest), and the scan itself is bounded. @@ -526,19 +597,12 @@ async fn list( truncated = true; break; } - // `Path` drops the trailing delimiter, so the remainder starts with one. - let Some(rel) = meta.location.as_ref().strip_prefix(prefix.as_ref()) else { - continue; - }; - let Some((sid, epoch)) = rel.trim_start_matches('/').split_once('/') else { - continue; - }; - let Ok(epoch) = epoch.parse::() else { - continue; - }; - if sid.is_empty() || sid.contains('/') { + if cutoff.is_some_and(|cutoff| meta.last_modified < cutoff) { continue; } + let Some((sid, IndexEntry::Marker(epoch))) = index_entry(&prefix, &meta.location) else { + continue; + }; newest.push(std::cmp::Reverse(( meta.last_modified, epoch, @@ -1206,15 +1270,29 @@ async fn push_session(backend: &Backend, s: &PushedSession) -> Result<(usize, bo Ok((written, false)) } -/// The marker goes first so a removal cut short leaves nothing listed, then the head so +/// The markers go first so a removal cut short leaves nothing listed, then the head so /// nothing pulls either, and no push takes it for a session still there (see `push_session`). +/// The retention sweep's record goes last (see `Backend::sweep_key`). async fn remove_session(backend: &Backend, sid: &str) -> Result<()> { + let sweep = backend.sweep_key(sid); backend - .delete_prefix(&backend.index_session_prefix(sid)) + .store + .list(Some(&backend.index_session_prefix(sid))) + .map_err(object_store_error_to_error) + .try_for_each_concurrent(IO_CONCURRENCY, |meta| { + let sweep = &sweep; + async move { + if meta.location == *sweep { + return Ok(()); + } + backend.delete(&meta.location).await + } + }) .await?; backend.delete(&backend.head_key(sid)).await?; backend.delete_prefix(&backend.session_prefix(sid)).await?; - backend.delete_prefix(&backend.images_prefix(sid)).await + backend.delete_prefix(&backend.images_prefix(sid)).await?; + backend.delete(&sweep).await } /// One writer per session at a time, across servers: a push and a removal of the same @@ -1261,6 +1339,205 @@ async fn remove_session_locked(db: &DB, backend: &Backend, sid: &str) -> Result< result } +/// Deletes, in every workspace with `ai_config.sessions_retention_days`, the backups of the +/// sessions whose marker is older than that: the marker is rewritten by every push that +/// completes, so its modification time is the session's last activity as the storage clocks +/// it. For the monitor, on every server: a session-level advisory lock keeps one pass at a +/// time across them. The walk reads markers only, one object per session and nothing of what +/// the sessions hold, under each user's prefix in turn (`list_with_delimiter` names the +/// users), and deletes at most `SWEEP_MAX_PER_WORKSPACE` sessions per workspace and pass. A +/// session goes under its lock (`lock_session`), once its markers are listed again there and +/// still all older (see `sweep_session`). A removal cut short leaves the sweep's record next +/// to the markers, which the walk also collects, so the next pass finishes it. +pub async fn sweep_expired_ai_session_backups(db: &DB) { + let mut lock_conn = match db.acquire().await { + Ok(conn) => conn, + Err(e) => { + tracing::error!("AI session retention: could not acquire a connection: {e:#}"); + return; + } + }; + let locked: bool = match sqlx::query_scalar("SELECT pg_try_advisory_lock($1)") + .bind(SWEEP_LOCK_ID) + .fetch_one(&mut *lock_conn) + .await + { + Ok(locked) => locked, + Err(e) => { + tracing::error!("AI session retention: advisory lock failed: {e:#}"); + return; + } + }; + if !locked { + return; + } + if let Err(e) = sweep_workspaces(db).await { + tracing::error!("AI session retention sweep failed: {e:#}"); + } + if let Err(e) = sqlx::query("SELECT pg_advisory_unlock($1)") + .bind(SWEEP_LOCK_ID) + .execute(&mut *lock_conn) + .await + { + tracing::error!("AI session retention: advisory unlock failed: {e:#}"); + } +} + +async fn sweep_workspaces(db: &DB) -> Result<()> { + let workspaces = sqlx::query_as::<_, (String, Option, i64, bool)>( + "SELECT workspace_id, ai_config->'sessions_retention_days', ai_sessions_backup_generation, \ + large_file_storage IS NOT NULL \ + FROM workspace_settings \ + WHERE ai_config->'sessions_retention_days' IS NOT NULL", + ) + .fetch_all(db) + .await?; + for (w_id, retention, generation, has_storage) in workspaces { + let Some(days) = sessions_retention_days(retention.as_ref()) else { + continue; + }; + match sweep_workspace(db, &w_id, days, generation, has_storage).await { + Ok(0) => {} + Ok(deleted) => tracing::info!( + "AI session retention deleted {deleted} session backups of {w_id} older than {days} days" + ), + Err(e) => tracing::warn!("AI session retention sweep of {w_id}: {e:#}"), + } + } + Ok(()) +} + +/// `has_storage` comes from the row `generation` was read from, as in `backend`: the instance +/// store is swept only under a generation read while the workspace had no storage of its own. +async fn sweep_workspace( + db: &DB, + w_id: &str, + days: u32, + generation: i64, + has_storage: bool, +) -> Result { + let resolved = if has_storage { + primary_store(db, w_id).await? + } else { + fallback_store(db).await? + }; + let Some(resolved) = resolved else { + return Ok(0); + }; + let key = get_workspace_key(w_id, db).await?; + let (store, storage_id) = (resolved.store, resolved.storage_id); + let cutoff = retention_cutoff(days); + let root = ObjectPath::from(generation_prefix(w_id, generation)); + let users = store + .list_with_delimiter(Some(&root)) + .await + .map_err(object_store_error_to_error)? + .common_prefixes; + let mut deleted = 0; + for user_prefix in users { + let Some(user) = user_prefix.filename() else { + continue; + }; + // The sweep decrypts nothing; the cipher is only what a `Backend` is made of. + let backend = Backend { + store: store.clone(), + mc: crypt_from_key_with_suffix(&key, user), + prefix: user_prefix.to_string(), + storage_id: storage_id.clone(), + generation, + retention_days: Some(days), + fallback: resolved.fallback, + }; + let index = backend.index_prefix(); + let mut markers = backend.store.list(Some(&index)); + let mut expired = std::collections::BTreeSet::new(); + while let Some(meta) = markers.next().await { + let meta = meta.map_err(object_store_error_to_error)?; + let sid = match index_entry(&index, &meta.location) { + Some((sid, IndexEntry::Sweep)) => sid, + Some((sid, IndexEntry::Marker(_) | IndexEntry::Push)) + if meta.last_modified < cutoff => + { + sid + } + _ => continue, + }; + expired.insert(sid.to_string()); + if deleted + expired.len() >= SWEEP_MAX_PER_WORKSPACE { + break; + } + } + for sid in expired { + match sweep_session(db, &backend, &sid, cutoff).await { + Ok(true) => deleted += 1, + Ok(false) => {} + Err(e) => tracing::warn!( + "AI session retention left the backup of {sid} in {w_id} for the next pass: {e:#}" + ), + } + } + if deleted >= SWEEP_MAX_PER_WORKSPACE { + break; + } + } + Ok(deleted) +} + +/// True when the session was deleted. Under the session's lock its markers are listed again: +/// one a push renewed since the walk keeps the session. A session with none is left alone +/// while a push split over parts is between two of them (its token younger than the +/// retention) or it is gone, unless the sweep's record says a removal was started; an older +/// token is a split push a browser abandoned, whose landed parts nothing lists. The record is +/// written before anything is deleted and removed last, so a removal cut short is found again +/// by the next pass. +async fn sweep_session( + db: &DB, + backend: &Backend, + sid: &str, + cutoff: chrono::DateTime, +) -> Result { + let tx = lock_session(db, backend, sid).await?; + let result = async { + let (sweep, push) = (backend.sweep_key(sid), backend.push_key(sid)); + let mut entries = backend.store.list(Some(&backend.index_session_prefix(sid))); + let (mut listed, mut renewed, mut started, mut abandoned) = (false, false, false, false); + while let Some(meta) = entries.next().await { + let meta = meta.map_err(object_store_error_to_error)?; + if meta.location == sweep { + started = true; + } else if meta.location == push { + abandoned = meta.last_modified < cutoff; + } else { + listed = true; + renewed |= meta.last_modified >= cutoff; + } + } + if renewed { + // A push listed the session again over a removal cut short before its markers + // went, which had deleted nothing else. + if started { + backend.delete(&sweep).await?; + } + return Ok(false); + } + if !listed && !started && !abandoned { + return Ok(false); + } + if !started { + backend + .store + .put(&sweep, PutPayload::new()) + .await + .map_err(object_store_error_to_error)?; + } + remove_session(backend, sid).await?; + Ok(true) + } + .await; + tx.commit().await?; + result +} + async fn push( authed: ApiAuthed, Extension(db): Extension, diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 715f0b6556..b70d80140a 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -71,6 +71,8 @@ mod ai_free_tier_ee; mod ai_free_tier_oss; #[cfg(feature = "parquet")] mod ai_sessions; +#[cfg(feature = "parquet")] +pub use ai_sessions::sweep_expired_ai_session_backups; mod ai_shared_artifacts; mod apps; mod apps_raw_bundle; diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index 55f244c143..a0a27eab4c 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -108,6 +108,7 @@ async fn edit_copilot_config( } ai_config.validate_model_pricing()?; + ai_config.validate_sessions_retention()?; let mut tx = db.begin().await?; @@ -148,6 +149,7 @@ async fn edit_copilot_config( let workspace_has_config = ai_config.has_providers(); let copilot_disabled = ai_config.copilot_disabled; let sessions_storage_disabled = ai_config.sessions_storage_disabled; + let sessions_retention_days = ai_config.sessions_retention_days; let instance_ai_config = sqlx::query_scalar!("SELECT value FROM global_settings WHERE name = 'ai_config'") .fetch_optional(&db) @@ -176,6 +178,7 @@ async fn edit_copilot_config( }; effective_ai_config.copilot_disabled = copilot_disabled; effective_ai_config.sessions_storage_disabled = sessions_storage_disabled; + effective_ai_config.sessions_retention_days = sessions_retention_days; Ok(Json(EditCopilotConfigResponse { effective_ai_config, @@ -217,6 +220,9 @@ async fn get_copilot_info( let sessions_storage_disabled = workspace_ai_config .as_ref() .is_some_and(|c| c.0.sessions_storage_disabled); + let sessions_retention_days = workspace_ai_config + .as_ref() + .and_then(|c| c.0.sessions_retention_days); let instance_config = sqlx::query_scalar!("SELECT value FROM global_settings WHERE name = 'ai_config'") .fetch_optional(&db) @@ -242,6 +248,7 @@ async fn get_copilot_info( }; effective.copilot_disabled = copilot_disabled; effective.sessions_storage_disabled = sessions_storage_disabled; + effective.sessions_retention_days = sessions_retention_days; Ok(Json(effective)) } diff --git a/docs/ai-session-backups.md b/docs/ai-session-backups.md index 3e0c747b8a..9af2bd3a84 100644 --- a/docs/ai-session-backups.md +++ b/docs/ai-session-backups.md @@ -201,7 +201,8 @@ its pulls are done, and just before they do the whole family is listed again (me backups were off included, since a move from another device can land in a workspace between the first listings and the pulls; a family of one, with nowhere else for a copy to show up, is not): a session a later copy of which showed up elsewhere is left, with the family, for the -next time. Only a user-initiated `deleteSession` removes the backup; the next push from +next time. Only a user-initiated `deleteSession`, and the retention sweep below, remove the +backup; the next push from another device that still has the session is refused with `needs_whole` (nothing of it is written), its row goes stale without a backoff, and that device's next flush sends the session whole; the workspace-lifecycle @@ -235,6 +236,37 @@ old one; a session whose own parts were answered from different storages is not all. The listing a restore starts with runs the same check, so a storage switch is noticed at the first push after it or on the next page load, whichever comes first. +## Retention + +`ai_config.sessions_retention_days` (per workspace, in the AI settings; unset by default; +the `sessions_storage_disabled` pattern: no migration, carried by settings export and the +CLI; 1 to 3650) puts an age on backups, counted from the last push of the session that +completed. It applies to the backup only: a browser keeps its copy whatever the retention, +and a backup swept while a browser still has the session comes back once that browser writes +to it again (its incremental push is refused and goes whole). + +The server sweeps the object store (`sweep_expired_ai_session_backups`, from the monitor about +every 40 minutes on each server, one pass at a time under a session-level advisory lock). For +every workspace with a retention it takes the store its backups live in, its own storage or +the instance store standing in, decided from the row it reads the generation from as the +routes do, names the users under the generation prefix (`list_with_delimiter`) and lists each +user's `index/` once: one object per session, nothing of what the sessions hold. A session +whose marker is older than the retention is removed under its lock (`lock_session`), once its +markers, listed again there, are still all older: a push that renewed the session between the +walk and the lock keeps it, and one split over parts either holds the lock or has the session +unlisted with its token next to the markers (`index/{sid}/push`), which the sweep leaves +alone while the token is younger than the retention: an older one is a push a browser +abandoned, whose landed parts nothing lists, and it goes the same way. Before deleting +anything the sweep writes a record next to the markers (`index/{sid}/sweep`, not an epoch, so +neither `list` nor `pull` counts it), and `remove_session` deletes it last: a removal cut +short, its markers already gone, is found by the next pass and finished, unless a push listed +the session again first. At most 1000 sessions per workspace and pass; the rest wait for the +next. `list` leaves an expired marker out of its answer meanwhile, so a browser never restores +a session the sweep has not reached. The marker's modification time is the storage's clock and +the cutoff the server's. The sweep reaches only the backups the routes would: a deleted +workspace's stay in its storage, and so do those a workspace keeps in the instance store once +`ai_sessions_instance_storage_fallback` is set to false. + ## Limits Push bodies are packed to about 8 MB (UTF-8 bytes as sent), at most 100 entries, 200 removals and diff --git a/frontend/src/lib/components/workspaceSettings/AISettings.svelte b/frontend/src/lib/components/workspaceSettings/AISettings.svelte index 5bc7974af2..a1dcc76758 100644 --- a/frontend/src/lib/components/workspaceSettings/AISettings.svelte +++ b/frontend/src/lib/components/workspaceSettings/AISettings.svelte @@ -29,6 +29,7 @@ import AiUsagePanel from './AiUsagePanel.svelte' import { setCopilotInfo } from '$lib/aiStore' import { backupSettingsChanged } from '$lib/components/sessions/sessionMirror.svelte' + import TextInput from '../text_input/TextInput.svelte' import AIPromptsModal from '../settings/AIPromptsModal.svelte' import { Settings } from 'lucide-svelte' import { untrack } from 'svelte' @@ -81,6 +82,7 @@ let workspaceOverrideEditorOpened = $state(false) let copilotDisabled = $state(false) let sessionsStorageDisabled = $state(false) + let sessionsRetentionDays: number | undefined = $state(undefined) // --- Initial state for dirty tracking --- let initialAiProviders: Exclude = $state({}) @@ -93,6 +95,7 @@ let initialPrompts: Record = $state({}) let initialCopilotDisabled = $state(false) let initialSessionsStorageDisabled = $state(false) + let initialSessionsRetentionDays: number | undefined = $state(undefined) let lastLoadedConfigKey = $state(undefined) function clone(v: T): T { @@ -122,6 +125,7 @@ modelPricing = clone(config?.model_pricing ?? {}) copilotDisabled = config?.copilot_disabled === true sessionsStorageDisabled = config?.sessions_storage_disabled === true + sessionsRetentionDays = config?.sessions_retention_days for (const mode of ['edit', 'fix', 'gen']) { if (!(mode in customPrompts)) { customPrompts[mode] = '' @@ -140,6 +144,7 @@ initialPrompts = clone(customPrompts) initialCopilotDisabled = copilotDisabled initialSessionsStorageDisabled = sessionsStorageDisabled + initialSessionsRetentionDays = sessionsRetentionDays } export function loadFromConfig(config: AIConfig | undefined) { @@ -157,6 +162,7 @@ modelPricing = clone(initialModelPricing) copilotDisabled = initialCopilotDisabled sessionsStorageDisabled = initialSessionsStorageDisabled + sessionsRetentionDays = initialSessionsRetentionDays } $effect(() => { @@ -193,7 +199,8 @@ JSON.stringify(maxTokensPerModel) !== JSON.stringify(initialMaxTokensPerModel) || JSON.stringify(modelPricing) !== JSON.stringify(initialModelPricing) || copilotDisabled !== initialCopilotDisabled || - sessionsStorageDisabled !== initialSessionsStorageDisabled + sessionsStorageDisabled !== initialSessionsStorageDisabled || + sessionsRetentionDays !== initialSessionsRetentionDays ) $effect(() => { @@ -298,9 +305,11 @@ .filter(([_, prompt]) => prompt.trim().length > 0) .reduce((acc, [mode, prompt]) => ({ ...acc, [mode]: prompt }), {}) - // The flags are what a workspace on instance defaults still stores of its own. + // The flags and the retention are what a workspace on instance defaults still stores + // of its own. const copilot_disabled = copilotDisabled ? true : undefined const sessions_storage_disabled = sessionsStorageDisabled ? true : undefined + const sessions_retention_days = sessionsRetentionDays return Object.keys(aiProviders ?? {}).length > 0 ? { providers: aiProviders, @@ -312,16 +321,27 @@ Object.keys(maxTokensPerModel).length > 0 ? maxTokensPerModel : undefined, model_pricing: Object.keys(modelPricing).length > 0 ? modelPricing : undefined, copilot_disabled, - sessions_storage_disabled + sessions_storage_disabled, + sessions_retention_days } - : { copilot_disabled, sessions_storage_disabled } + : { copilot_disabled, sessions_storage_disabled, sessions_retention_days } } + // The server refuses anything outside this range; the input holds the same bounds. + const MAX_SESSIONS_RETENTION_DAYS = 3650 + let retentionInvalid = $derived( + sessionsRetentionDays !== undefined && + (!Number.isInteger(sessionsRetentionDays) || + sessionsRetentionDays < 1 || + sessionsRetentionDays > MAX_SESSIONS_RETENTION_DAYS) + ) + function isSaveDisabled(): boolean { return ( !Object.values(aiProviders).every((p) => p.resource_path) || (metadataModel != undefined && metadataModel.length === 0) || - (codeCompletionModel != undefined && codeCompletionModel.length === 0) + (codeCompletionModel != undefined && codeCompletionModel.length === 0) || + retentionInvalid ) } @@ -671,6 +691,33 @@ options={{ right: 'Do not back AI sessions up to the workspace storage' }} /> + +
+
+ sessionsRetentionDays ?? '', + (v) => { + const n = typeof v === 'number' ? v : parseInt(v ?? '') + sessionsRetentionDays = Number.isNaN(n) ? undefined : n + } + } + /> +
+ days +
+ {/if}