diff --git a/backend/.sqlx/query-c2f3492c2d80f5c6d157d8c1dab7f9ed1c0f4b4d1f07789b40945cfb8a3a7b39.json b/backend/.sqlx/query-c2f3492c2d80f5c6d157d8c1dab7f9ed1c0f4b4d1f07789b40945cfb8a3a7b39.json new file mode 100644 index 0000000000..9724a99b18 --- /dev/null +++ b/backend/.sqlx/query-c2f3492c2d80f5c6d157d8c1dab7f9ed1c0f4b4d1f07789b40945cfb8a3a7b39.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE workspace_settings SET ai_sessions_backup_generation = ai_sessions_backup_generation + 1 WHERE workspace_id = $1 AND large_file_storage IS NULL", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [] + }, + "hash": "c2f3492c2d80f5c6d157d8c1dab7f9ed1c0f4b4d1f07789b40945cfb8a3a7b39" +} diff --git a/backend/.sqlx/query-dfa82a3f291cdc8f05cc4af6114e67c90e86d35dd5f74b1366f802dbb951ef9a.json b/backend/.sqlx/query-dfa82a3f291cdc8f05cc4af6114e67c90e86d35dd5f74b1366f802dbb951ef9a.json new file mode 100644 index 0000000000..0b17793580 --- /dev/null +++ b/backend/.sqlx/query-dfa82a3f291cdc8f05cc4af6114e67c90e86d35dd5f74b1366f802dbb951ef9a.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT large_file_storage IS NOT NULL AS \"has_storage!\" FROM workspace_settings WHERE workspace_id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "has_storage!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "dfa82a3f291cdc8f05cc4af6114e67c90e86d35dd5f74b1366f802dbb951ef9a" +} diff --git a/backend/.sqlx/query-ea7bc2e5f53144ca23f8b3dea71cb6a994e969d915d50afbf0cb8547e6decf11.json b/backend/.sqlx/query-ea7bc2e5f53144ca23f8b3dea71cb6a994e969d915d50afbf0cb8547e6decf11.json new file mode 100644 index 0000000000..a344e64392 --- /dev/null +++ b/backend/.sqlx/query-ea7bc2e5f53144ca23f8b3dea71cb6a994e969d915d50afbf0cb8547e6decf11.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT ai_sessions_backup_generation FROM workspace_settings WHERE workspace_id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "ai_sessions_backup_generation", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "ea7bc2e5f53144ca23f8b3dea71cb6a994e969d915d50afbf0cb8547e6decf11" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 7acd297dac..5d28f73b5a 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -1c1dab33563c4907aff8b0da825fb66db60af82a +93433d7c9dc34f2c0f56a5297d3453aabd2f9472 \ No newline at end of file diff --git a/backend/tests/ai_sessions.rs b/backend/tests/ai_sessions.rs index 340a1380da..3eb28cfae8 100644 --- a/backend/tests/ai_sessions.rs +++ b/backend/tests/ai_sessions.rs @@ -34,6 +34,54 @@ async fn configure_primary_lfs(db: &Pool, root_path: &str) -> anyhow:: Ok(()) } +/// Configures the primary storage through the route, which is what sweeps the workspace's +/// backups out of the instance store. +async fn configure_primary_lfs_via_route(base: &str, root_path: &str) -> anyhow::Result<()> { + let resp = authed( + client().post(format!("{base}/workspaces/edit_large_file_storage_config")), + "SECRET_TOKEN", + ) + .json(&json!({ "large_file_storage": { + "type": "FilesystemStorage", + "root_path": root_path, + "public_resource": false, + "advanced_permissions": null, + "secondary_storage": {} + }})) + .send() + .await?; + assert_eq!(resp.status(), 200, "{}", resp.text().await?); + Ok(()) +} + +/// The instance setting allowing the instance store to stand in for a workspace without +/// storage: `None` leaves it unset, which is on. +async fn set_instance_fallback(db: &Pool, on: Option) -> anyhow::Result<()> { + sqlx::query("DELETE FROM global_settings WHERE name = 'ai_sessions_instance_storage_fallback'") + .execute(db) + .await?; + if let Some(on) = on { + sqlx::query( + "INSERT INTO global_settings (name, value) VALUES ('ai_sessions_instance_storage_fallback', $1)", + ) + .bind(json!(on)) + .execute(db) + .await?; + } + Ok(()) +} + +/// Polls until nothing is under the directory, for a deletion that runs off the request. +async fn wait_until_empty(dir: &std::path::Path, what: &str) { + for _ in 0..100 { + if files_under(dir).is_empty() { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + panic!("{what}: objects left under {}", dir.display()); +} + async fn list(base: &str, token: &str) -> anyhow::Result { let resp = authed(client().get(format!("{base}/ai/sessions/list")), token) .send() @@ -143,7 +191,9 @@ async fn test_backups_round_trip_encrypted_and_scoped_to_the_user( server.addr.port() ); - // No storage configured: the browser is told to stop trying. + // No storage configured, and the instance store (another test of this process may + // have loaded one) not allowed to stand in: the browser is told to stop trying. + set_instance_fallback(&db, Some(false)).await?; let listing = list(&base, "SECRET_TOKEN").await?; assert_eq!(listing["enabled"], false); assert_eq!(listing["sessions"], json!([])); @@ -1184,3 +1234,177 @@ async fn test_backup_writes_are_refused_for_the_wrong_owner_token_or_id( assert!(files_under(storage_dir.path()).is_empty()); Ok(()) } + +/// Puts the process-wide instance store back to none, even when an assertion fails. +struct ResetInstanceStore; +impl Drop for ResetInstanceStore { + fn drop(&mut self) { + if let Ok(mut store) = windmill_object_store::OBJECT_STORE_SETTINGS.try_write() { + *store = None; + } + } +} + +/// Puts the process-wide license key id back to none, an Enterprise plan in this build, +/// even when an assertion fails. +struct ResetLicensePlan; +impl Drop for ResetLicensePlan { + fn drop(&mut self) { + windmill_common::ee::LICENSE_KEY_ID.store(std::sync::Arc::new(String::new())); + } +} + +/// A workspace without storage of its own backs up to the instance object store, every +/// answer saying so (`fallback`); a storage of its own, once configured, answers instead, +/// under a generation past everything the workspace left in the instance store, which the +/// change deletes; a plan switched to Pro stops the fallback with the store still loaded. +#[sqlx::test(fixtures("base"))] +async fn test_backups_fall_back_to_the_instance_storage(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() + ); + + // The instance store, built from its settings as `reload_object_store_setting` does. + let instance_dir = tempfile::tempdir()?; + let instance_root = instance_dir.path().to_string_lossy().to_string(); + *windmill_object_store::OBJECT_STORE_SETTINGS.write().await = Some( + windmill_object_store::build_object_store_from_settings( + windmill_object_store::ObjectSettings::Filesystem( + windmill_object_store::FilesystemSettings { root_path: instance_root.clone() }, + ), + None, + ) + .await?, + ); + let _reset = ResetInstanceStore; + let in_instance = instance_dir + .path() + .join("windmill_ai_sessions/test-workspace"); + + // Turned off by the instance setting: the browser is told to stop trying. + set_instance_fallback(&db, Some(false)).await?; + assert_eq!(list(&base, "SECRET_TOKEN").await?["enabled"], false); + set_instance_fallback(&db, None).await?; + + // On, as it is unless turned off: the backups land in the instance store, under the + // workspace's prefix, and every answer says which kind of store it came from. + let head = + json!({ "id": "s1", "workspace_id": "test-workspace", "createdAt": 1, "chatId": "c1" }); + let entry = json!({ "id": "s1", "whole": true, "head": head, "chats": [{ "id": "c1", "record": { "id": "c1" } }] }); + let push_whole = || { + push( + &base, + "SECRET_TOKEN", + json!({ "owner": "test@windmill.dev", "sessions": [entry.clone()] }), + ) + }; + let resp = push_whole().await?; + assert_eq!(resp.status(), 200, "{}", resp.text().await?); + let pushed: Value = resp.json().await?; + assert_eq!(pushed["fallback"], true); + assert_eq!(pushed["results"], json!([{ "id": "s1" }])); + let listing = list(&base, "SECRET_TOKEN").await?; + assert_eq!(listing["enabled"], true); + assert_eq!(listing["fallback"], true); + assert_eq!(listing["sessions"][0]["id"], "s1"); + let fallback_storage_id = listing["storage_id"].clone(); + let pulled = pull(&base, "SECRET_TOKEN", &["s1"]).await?; + assert_eq!(pulled["fallback"], true); + assert_eq!(pulled["sessions"][0]["head"], head); + assert!(!files_under(&in_instance).is_empty()); + + // The workspace's storage usage counts them, under a name of their own. + let resp = authed( + client().get(format!("{base}/job_helpers/storage_usage")), + "SECRET_TOKEN", + ) + .send() + .await?; + assert_eq!(resp.status(), 200, "{}", resp.text().await?); + let usage: Value = resp.json().await?; + let fallback_usage = usage["storages"] + .as_array() + .unwrap() + .iter() + .find(|s| s["storage"] == "_ai_sessions_fallback_") + .unwrap_or_else(|| panic!("no fallback usage in {usage}")); + assert!(fallback_usage["bytes"].as_i64().unwrap() > 0); + + // 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; + assert_eq!(list(&base, "SECRET_TOKEN").await?["sessions"], json!([])); + assert_eq!(push_whole().await?.status(), 200); + assert!(!files_under(&in_instance).is_empty()); + + // A storage of its own answers instead, under a generation the configuration moved past + // everything the workspace left in the instance store: nothing there is read again, + // whichever store a later return to the fallback finds, and it is deleted. + let before = list(&base, "SECRET_TOKEN").await?; + let storage_dir = tempfile::tempdir()?; + configure_primary_lfs_via_route(&base, &storage_dir.path().to_string_lossy()).await?; + wait_until_empty(&in_instance, "configuring a workspace storage").await; + let listing = list(&base, "SECRET_TOKEN").await?; + assert_eq!(listing["enabled"], true); + assert!(listing.get("fallback").is_none(), "{listing}"); + assert_ne!(listing["storage_id"], fallback_storage_id); + assert_eq!( + listing["backup_generation"].as_i64(), + before["backup_generation"].as_i64().map(|g| g + 1), + "configuring a storage over the fallback must move the generation on" + ); + assert_eq!(listing["sessions"], json!([])); + assert_eq!(push_whole().await?.status(), 200); + assert!(!files_under(storage_dir.path()).is_empty()); + assert!(files_under(&in_instance).is_empty()); + let resp = authed( + client().get(format!("{base}/job_helpers/storage_usage?refresh=true")), + "SECRET_TOKEN", + ) + .send() + .await?; + assert_eq!(resp.status(), 200, "{}", resp.text().await?); + let usage: Value = resp.json().await?; + assert!( + !usage.to_string().contains("_ai_sessions_fallback_"), + "nothing is counted in the instance store for a workspace with storage: {usage}" + ); + + // Pointed at the instance store's own bucket, a storage of its own keeps its live + // backups there under the current generation, which no storage change deletes. + configure_primary_lfs_via_route(&base, &instance_root).await?; + assert_eq!(push_whole().await?.status(), 200); + assert!(!files_under(&in_instance).is_empty()); + let same = list(&base, "SECRET_TOKEN").await?; + configure_primary_lfs_via_route(&base, &instance_root).await?; + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + assert!(!files_under(&in_instance).is_empty()); + let listing = list(&base, "SECRET_TOKEN").await?; + assert!(listing.get("fallback").is_none(), "{listing}"); + assert_eq!(listing["backup_generation"], same["backup_generation"]); + assert_eq!(listing["sessions"][0]["id"], "s1"); + + // Back to no storage of its own, the fallback answers; a plan switched to Pro while the + // instance store stays loaded stops it at once, for the listing and the push alike. + let resp = authed( + client().post(format!("{base}/workspaces/edit_large_file_storage_config")), + "SECRET_TOKEN", + ) + .json(&json!({ "large_file_storage": null })) + .send() + .await?; + assert_eq!(resp.status(), 200, "{}", resp.text().await?); + assert_eq!(list(&base, "SECRET_TOKEN").await?["fallback"], true); + let _enterprise_again = ResetLicensePlan; + windmill_common::ee::LICENSE_KEY_ID.store(std::sync::Arc::new("test_pro".to_string())); + assert_eq!(list(&base, "SECRET_TOKEN").await?["enabled"], false); + let resp = push_whole().await?; + assert_eq!(resp.status(), 200, "{}", resp.text().await?); + let pushed: Value = resp.json().await?; + assert_eq!(pushed["enabled"], false, "{pushed}"); + + Ok(()) +} diff --git a/backend/windmill-api-workspaces/src/ai_session_backups.rs b/backend/windmill-api-workspaces/src/ai_session_backups.rs index bbdbfbc8c9..46c91d5354 100644 --- a/backend/windmill-api-workspaces/src/ai_session_backups.rs +++ b/backend/windmill-api-workspaces/src/ai_session_backups.rs @@ -1,5 +1,6 @@ -//! What the workspace key rotation and the AI session backup routes -//! (`windmill-api/src/ai_sessions.rs`) share about the backups in the workspace storage. +//! What the workspace key rotation, the workspace storage settings and the AI session backup +//! routes (`windmill-api/src/ai_sessions.rs`) share about the backups: the store they live +//! in, and what a rotation or a storage change deletes. //! //! The backups are ciphertext under the workspace key and live under a prefix named by a //! generation the rotation bumps (`workspace_settings.ai_sessions_backup_generation`) in the @@ -13,17 +14,29 @@ //! one to read with and nothing to rewrite in place. A generation is never reused, so no //! deletion, however late, can touch live objects; a rotation that fails before its commit //! bumps nothing and deletes nothing; two rotations racing serialize on the key row. +//! +//! A workspace without storage of its own keeps its backups in the instance object store +//! instead, under the same layout and key, while `ai_sessions_instance_storage_fallback` +//! allows it. Configuring a storage for such a workspace bumps the generation in the +//! transaction that sets it, so everything the workspace left in any instance store sits +//! under a generation the routes never read again: a later return to the instance store, +//! whichever it is by then, starts from a newer one. That is what lets a storage change +//! delete the older generations from the instance store without fencing against what +//! happens next, and a browser retire a removal owed to an instance store once the +//! workspace's own storage answered. use std::sync::Arc; -use futures::TryStreamExt; +use futures::{StreamExt, TryStreamExt}; use windmill_common::error::{Error, Result}; use windmill_common::utils::calculate_hash; use windmill_common::DB; use windmill_object_store::object_store_reexports::{ ObjectStore, ObjectStoreError, Path as ObjectPath, }; -use windmill_object_store::{object_store_error_to_error, ObjectStoreResource}; +use windmill_object_store::{ + object_store_error_to_error, object_store_location, ObjectStoreResource, +}; use windmill_types::s3::LargeFileStorage; /// The root of every AI session backup key in a workspace's storage. @@ -31,6 +44,9 @@ pub const ROOT: &str = "windmill_ai_sessions"; /// The push body cap: no object written through the routes is larger. One that is was /// planted by whoever holds the bucket's credentials, and is left unread. pub const MAX_OBJECT_BYTES: usize = 32 * 1024 * 1024; +/// The storage name the workspace's backups in the instance store count under in its +/// storage usage, next to `_default_` and the secondary storages. +pub const FALLBACK_STORAGE: &str = "_ai_sessions_fallback_"; const IO_CONCURRENCY: usize = 8; @@ -39,34 +55,76 @@ pub fn generation_prefix(w_id: &str, generation: i64) -> String { format!("{ROOT}/{w_id}/g{generation}") } +/// The prefix of everything the workspace ever backed up, whatever the generation. +fn workspace_prefix(w_id: &str) -> ObjectPath { + ObjectPath::from(format!("{ROOT}/{w_id}")) +} + /// Names the storage the backups are in, by what locates its objects (endpoint, region, /// bucket; never the credentials, which rotate), so a browser tells that its sync state was /// recorded against another storage; the generation, answered alongside, tells it a /// rotation happened in this one. pub fn storage_id(resource: &ObjectStoreResource) -> String { - let location = match resource { - ObjectStoreResource::S3(s) => format!( - "s3:{}:{}:{}:{}", - s.endpoint, - s.port.unwrap_or_default(), - s.region, - s.bucket - ), - ObjectStoreResource::Azure(a) => format!( - "azure:{}:{}:{}", - a.endpoint.as_deref().unwrap_or_default(), - a.account_name, - a.container_name - ), - ObjectStoreResource::Gcs(g) => format!("gcs:{}", g.bucket), - ObjectStoreResource::Filesystem(f) => format!("fs:{}", f.root_path), - }; - calculate_hash(&location)[..16].to_string() + calculate_hash(&object_store_location(resource))[..16].to_string() } -/// The workspace's primary storage, resolved without a caller: a rotation runs the -/// deletion off its own request. -async fn primary_store(db: &DB, w_id: &str) -> Result>> { +/// Where a workspace's backups live: its primary storage, or the instance object store +/// standing in for it. +pub struct BackupStore { + pub store: Arc, + pub storage_id: String, + pub fallback: bool, +} + +/// The instance object store, for a workspace without storage of its own: loaded from +/// settings that say where its objects are, and not turned off by +/// `ai_sessions_instance_storage_fallback`, which is on unless set to false. Named like a +/// workspace storage, by that location, in a namespace of its own. Never on the Pro plan, +/// checked on every call: a store loaded before a switch to Pro stays loaded. Never in a +/// build without `private`, which has neither workspace storage nor the quota the fallback +/// counts toward. +/// +/// Authorizes nothing, and the store reaches every workspace's objects: the caller must have +/// authorized the user for the workspace and keep what it reads and writes under that +/// user's prefix in it, as the backup routes do. +pub async fn fallback_store(db: &DB) -> Result> { + #[cfg(not(feature = "private"))] + { + let _ = db; + Ok(None) + } + #[cfg(feature = "private")] + { + if matches!( + windmill_common::ee_oss::get_license_plan().await, + windmill_common::ee_oss::LicensePlan::Pro + ) { + return Ok(None); + } + let Some((store, Some(location))) = + windmill_object_store::get_object_store_with_location().await + else { + return Ok(None); + }; + let setting = windmill_common::global_settings::load_value_from_global_settings( + db, + windmill_common::global_settings::AI_SESSIONS_INSTANCE_STORAGE_FALLBACK_SETTING, + ) + .await?; + if matches!(setting, Some(serde_json::Value::Bool(false))) { + return Ok(None); + } + Ok(Some(BackupStore { + storage_id: calculate_hash(&format!("instance:{location}"))[..16].to_string(), + store, + fallback: true, + })) + } +} + +/// 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> { let Some(lfs_json) = sqlx::query_scalar!( "SELECT large_file_storage FROM workspace_settings WHERE workspace_id = $1", w_id @@ -91,9 +149,19 @@ async fn primary_store(db: &DB, w_id: &str) -> Result Result> { + if let Some(primary) = primary_store(db, w_id).await? { + return Ok(Some(primary)); + } + fallback_store(db).await } /// The generation an object key sits under, `None` for a key of no generation (an older @@ -107,40 +175,41 @@ fn generation_of(w_id: &str, key: &ObjectPath) -> Option { .ok() } -/// Deletes, off the request and as the listing streams, every object of the workspace's -/// backups from a generation older than `current`, once the rotation that made `current` -/// the generation has committed: nothing writes there any more but a push that resolved its -/// prefix before the commit, junk the browser's next push of that session rewrites under the -/// current prefix, as is anything a deletion cut short left behind. For the rotation route, -/// which authorized its caller as a superadmin. +/// Deletes, as the listing streams, every object of the workspace's backups in the store +/// from a generation older than `current`. +async fn delete_older(store: &Arc, w_id: &str, current: i64) -> Result<()> { + store + .list(Some(&workspace_prefix(w_id))) + .map_err(object_store_error_to_error) + .try_for_each_concurrent(IO_CONCURRENCY, |meta| async move { + if generation_of(w_id, &meta.location).is_some_and(|g| g >= current) { + return Ok(()); + } + match store.delete(&meta.location).await { + Ok(()) | Err(ObjectStoreError::NotFound { .. }) => Ok(()), + Err(e) => Err(object_store_error_to_error(e)), + } + }) + .await +} + +/// Deletes, off the request, every object of the workspace's backups from a generation +/// older than `current`, once the rotation that made `current` the generation has +/// committed: nothing writes there any more but a push that resolved its prefix before the +/// commit, junk the browser's next push of that session rewrites under the current prefix, +/// as is anything a deletion cut short left behind. For the rotation route, which +/// authorized its caller as a superadmin. pub(crate) fn spawn_delete_older(db: DB, w_id: String, current: i64) { tokio::spawn(async move { - let store = match primary_store(&db, &w_id).await { - Ok(Some(store)) => store, + let store = match workspace_store(&db, &w_id).await { + Ok(Some(store)) => store.store, Ok(None) => return, Err(e) => { tracing::warn!("older AI session backups of {w_id} left in place: {e:#}"); return; } }; - let prefix = ObjectPath::from(format!("{ROOT}/{w_id}")); - let deleted = store - .list(Some(&prefix)) - .map_err(object_store_error_to_error) - .try_for_each_concurrent(IO_CONCURRENCY, |meta| { - let (store, w_id) = (&store, &w_id); - async move { - if generation_of(w_id, &meta.location).is_some_and(|g| g >= current) { - return Ok(()); - } - match store.delete(&meta.location).await { - Ok(()) | Err(ObjectStoreError::NotFound { .. }) => Ok(()), - Err(e) => Err(object_store_error_to_error(e)), - } - } - }) - .await; - match deleted { + match delete_older(&store, &w_id, current).await { Ok(()) => { tracing::info!("deleted the AI session backups of {w_id} older than g{current}") } @@ -148,3 +217,57 @@ pub(crate) fn spawn_delete_older(db: DB, w_id: String, current: i64) { } }); } + +/// Deletes, off the request, what the workspace's backups left in the instance store under +/// a generation older than `current`, the one a storage settings change committed. Nothing +/// reads there: the routes use the workspace's own storage, or, back in the instance store, +/// `current` or a newer generation, since configuring a storage over the fallback bumped +/// it. So it runs whatever the storage is now and whatever the setting says (copies from +/// when it was on may be there), and a deletion that is slow, cut short or overtaken by a +/// later change deletes nothing live. For the storage settings route, which authorized its +/// caller as a workspace admin. +pub(crate) fn spawn_delete_fallback(w_id: String, current: i64) { + tokio::spawn(async move { + let Some(instance) = windmill_object_store::get_object_store().await else { + return; + }; + match delete_older(&instance, &w_id, current).await { + Ok(()) => tracing::info!( + "deleted the AI session backups of {w_id} older than g{current} from the instance store" + ), + Err(e) => tracing::warn!( + "deleting the AI session backups of {w_id} from the instance store: {e:#}" + ), + } + }); +} + +/// The bytes of the workspace's backups in the instance store, for its storage usage while +/// it has no storage of its own (once it has one nothing writes there, and the change +/// deleted what was): `None` when it has one, when there is no instance store, or when +/// there is nothing, so no empty usage entry shows up. Whether the setting is on or off, +/// since copies from when it was on may be there. +/// +/// Authorizes nothing: for the storage usage recount, which reports a total for the +/// workspace it was run for and hands out nothing it read. +pub async fn fallback_bytes(db: &DB, w_id: &str) -> Result> { + let has_storage = sqlx::query_scalar!( + r#"SELECT large_file_storage IS NOT NULL AS "has_storage!" FROM workspace_settings WHERE workspace_id = $1"#, + w_id + ) + .fetch_optional(db) + .await? + .unwrap_or(false); + if has_storage { + return Ok(None); + } + let Some(instance) = windmill_object_store::get_object_store().await else { + return Ok(None); + }; + let mut total: i64 = 0; + let mut stream = instance.list(Some(&workspace_prefix(w_id))); + while let Some(meta) = stream.next().await { + total += meta.map_err(object_store_error_to_error)?.size as i64; + } + Ok((total > 0).then_some(total)) +} diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index 1a3c24f096..c240948105 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -2093,6 +2093,17 @@ async fn edit_large_file_storage_config( serde_json::to_value::(lfs_config) .map_err(|err| Error::internal_err(err.to_string()))?; + // A workspace whose AI session backups fell back to the instance store leaves it + // here: the generation moves on, so nothing it left in any instance store is read + // again, whichever one a later return to the fallback finds (`ai_session_backups`). + sqlx::query!( + "UPDATE workspace_settings SET ai_sessions_backup_generation = \ + ai_sessions_backup_generation + 1 \ + WHERE workspace_id = $1 AND large_file_storage IS NULL", + &w_id + ) + .execute(&mut *tx) + .await?; sqlx::query!( "UPDATE workspace_settings SET large_file_storage = $1 WHERE workspace_id = $2", serialized_lfs_config, @@ -2108,8 +2119,23 @@ async fn edit_large_file_storage_config( .execute(&mut *tx) .await?; } + let backups_generation = sqlx::query_scalar!( + "SELECT ai_sessions_backup_generation FROM workspace_settings WHERE workspace_id = $1", + &w_id + ) + .fetch_optional(&mut *tx) + .await?; tx.commit().await?; + // Read by nothing any more, whatever the storage is now: what the AI session backups + // left in the instance store under a generation older than the one just committed. + #[cfg(feature = "parquet")] + if let Some(generation) = backups_generation { + crate::ai_session_backups::spawn_delete_fallback(w_id.clone(), generation); + } + #[cfg(not(feature = "parquet"))] + let _ = backups_generation; + // Trigger git sync for large file storage changes handle_deployment_metadata( &authed.email, diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 9f742a129b..63e8dd46ea 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -13222,6 +13222,9 @@ paths: backup_generation: type: integer description: bumped by every workspace key rotation; sync state recorded under another one is void + fallback: + type: boolean + description: the storage answered from is the instance object store, standing in for a workspace without storage of its own; a removal owed to it is retired by any answer from the workspace's own storage sessions: type: array description: the newest 500 at most @@ -13273,6 +13276,8 @@ paths: type: string backup_generation: type: integer + fallback: + type: boolean sessions: type: array items: @@ -13327,6 +13332,8 @@ paths: type: string backup_generation: type: integer + fallback: + type: boolean results: type: array items: diff --git a/backend/windmill-api/src/ai_sessions.rs b/backend/windmill-api/src/ai_sessions.rs index fca9c35fca..776b3a0561 100644 --- a/backend/windmill-api/src/ai_sessions.rs +++ b/backend/windmill-api/src/ai_sessions.rs @@ -30,7 +30,7 @@ use serde_json::value::RawValue; use std::sync::Arc; use windmill_api_auth::is_effectively_unscoped; use windmill_api_workspaces::ai_session_backups::{ - generation_prefix, storage_id, MAX_OBJECT_BYTES, + fallback_store, generation_prefix, storage_id, MAX_OBJECT_BYTES, }; use windmill_common::error::{Error, JsonResult, Result}; use windmill_common::utils::calculate_hash; @@ -101,6 +101,9 @@ struct Backend { /// owed to the storage alone (a rotation deleted the older generation's copy anyway). storage_id: String, generation: i64, + /// The store is the instance object store standing in for a workspace without storage + /// of its own (`ai_session_backups::fallback_store`). + fallback: bool, } impl Backend { @@ -408,34 +411,56 @@ fn require_json_object(kind: &str, raw: &RawValue, max_bytes: usize) -> Result<( Ok(()) } -/// `None` when the workspace has nowhere to keep backups: no primary storage configured, or -/// the admin switched them off. Both read as `enabled: false` so the browser stops trying. +/// `None` when the workspace has nowhere to keep backups: no primary storage configured and +/// 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) = sqlx::query_as::<_, (Option, i64)>( - "SELECT (ai_config->>'sessions_storage_disabled')::bool, ai_sessions_backup_generation \ - FROM workspace_settings WHERE workspace_id = $1", + 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)); + .unwrap_or((None, 0, false)); if disabled.unwrap_or(false) { return Ok(None); } - let (_, resource) = - crate::job_helpers_oss::get_workspace_s3_resource(authed, db, None, w_id, None).await?; - let Some(resource) = resource else { - return Ok(None); + // 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`). + let (store, storage_id, fallback) = if has_storage { + let (_, resource) = + crate::job_helpers_oss::get_workspace_s3_resource(authed, db, None, w_id, None).await?; + let Some(resource) = resource else { + return Ok(None); + }; + ( + build_object_store_client(&resource).await?, + storage_id(&resource), + false, + ) + } else { + // The instance store stands in, under the same layout and the same key. + match fallback_store(db).await? { + Some(f) => (f.store, f.storage_id, true), + None => return Ok(None), + } }; - let store = build_object_store_client(&resource).await?; let user = calculate_hash(&authed.email); // Keyed per user, not per workspace: anyone who can write the bucket could otherwise copy // another member's ciphertext under their own prefix and have `pull` decrypt it for them. let key = get_workspace_key(w_id, db).await?; let mc = crypt_from_key_with_suffix(&key, &user); - let storage_id = storage_id(&resource); let prefix = format!("{}/{user}", generation_prefix(w_id, generation)); - Ok(Some(Backend { store, mc, prefix, storage_id, generation })) + Ok(Some(Backend { + store, + mc, + prefix, + storage_id, + generation, + fallback, + })) } #[derive(Serialize)] @@ -455,6 +480,12 @@ struct ListResponse { storage_id: Option, #[serde(skip_serializing_if = "Option::is_none")] backup_generation: Option, + /// The storage is the instance store standing in for a workspace without one of its + /// own; a removal owed to it is retired by any answer from the workspace's own storage + /// once it has one (configuring it moved the generation past everything the workspace + /// left in any instance store). + #[serde(skip_serializing_if = "std::ops::Not::not")] + fallback: bool, sessions: Vec, /// The user has more sessions than the answer names. #[serde(skip_serializing_if = "std::ops::Not::not")] @@ -474,6 +505,7 @@ async fn list( enabled: false, storage_id: None, backup_generation: None, + fallback: false, sessions: vec![], truncated: false, })); @@ -526,6 +558,7 @@ async fn list( enabled: true, storage_id: Some(backend.storage_id.clone()), backup_generation: Some(backend.generation), + fallback: backend.fallback, sessions, truncated, })) @@ -594,6 +627,8 @@ struct PullResponse { storage_id: Option, #[serde(skip_serializing_if = "Option::is_none")] backup_generation: Option, + #[serde(skip_serializing_if = "std::ops::Not::not")] + fallback: bool, sessions: Vec, /// Ids that did not fit the response budget; ask for them again. deferred: Vec, @@ -839,6 +874,7 @@ async fn pull( enabled: false, storage_id: None, backup_generation: None, + fallback: false, sessions: vec![], deferred: vec![], })); @@ -861,6 +897,7 @@ async fn pull( enabled: true, storage_id: Some(backend.storage_id), backup_generation: Some(backend.generation), + fallback: backend.fallback, sessions, deferred, })) @@ -952,6 +989,8 @@ struct PushResponse { storage_id: Option, #[serde(skip_serializing_if = "Option::is_none")] backup_generation: Option, + #[serde(skip_serializing_if = "std::ops::Not::not")] + fallback: bool, results: Vec, } @@ -1241,6 +1280,7 @@ async fn push( enabled: false, storage_id: None, backup_generation: None, + fallback: false, results: vec![], })); }; @@ -1294,16 +1334,16 @@ async fn push( results.push(PushResult { id: sid.clone(), error, needs_whole: false }); } // Overwrites and deletes make this an over-count; the periodic recount the quota check - // schedules once usage is stale settles it. + // schedules once usage is stale settles it. Bytes in the instance store count under a + // name of their own, which the recount lists there. #[cfg(not(feature = "enterprise"))] if written > 0 { - crate::job_helpers_oss::bump_storage_usage( - &db, - &w_id, - windmill_object_store::DEFAULT_STORAGE, - written as i64, - ) - .await; + let storage = if backend.fallback { + windmill_api_workspaces::ai_session_backups::FALLBACK_STORAGE + } else { + windmill_object_store::DEFAULT_STORAGE + }; + crate::job_helpers_oss::bump_storage_usage(&db, &w_id, storage, written as i64).await; } #[cfg(feature = "enterprise")] let _ = written; @@ -1311,6 +1351,7 @@ async fn push( enabled: true, storage_id: Some(backend.storage_id), backup_generation: Some(backend.generation), + fallback: backend.fallback, results, })) } diff --git a/backend/windmill-common/src/global_settings.rs b/backend/windmill-common/src/global_settings.rs index 49efa26f60..ef63fc2347 100644 --- a/backend/windmill-common/src/global_settings.rs +++ b/backend/windmill-common/src/global_settings.rs @@ -84,6 +84,11 @@ pub const SANDBOX_REGISTRY_AUTH_SETTING: &str = "sandbox_registry_auth"; // windmill-worker/src/ssh_executor_ee.rs. pub const SSH_EXECUTION_SETTING: &str = "ssh_execution_enabled"; pub const OBJECT_STORE_CONFIG_SETTING: &str = "object_store_cache_config"; +/// Whether the instance object store stands in for a workspace without storage of its own +/// as the place its members' AI sessions are backed up to. On unless the row says `false`; +/// inert without an instance object store. +pub const AI_SESSIONS_INSTANCE_STORAGE_FALLBACK_SETTING: &str = + "ai_sessions_instance_storage_fallback"; /// Compile a newly deployed script's binary right after its dependency job and push it /// to the instance object store, so the first run does not pay the compile. Inert unless /// instance object storage is configured — without it the binary would only ever land in diff --git a/backend/windmill-common/src/instance_config.rs b/backend/windmill-common/src/instance_config.rs index 1dd3396809..de3fd7684a 100644 --- a/backend/windmill-common/src/instance_config.rs +++ b/backend/windmill-common/src/instance_config.rs @@ -264,6 +264,8 @@ pub struct GlobalSettings { pub disable_hub: Option, #[serde(skip_serializing_if = "Option::is_none")] pub auto_build_binary_on_deploy: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub ai_sessions_instance_storage_fallback: Option, // String settings #[serde(skip_serializing_if = "Option::is_none")] diff --git a/backend/windmill-object-store/src/lib.rs b/backend/windmill-object-store/src/lib.rs index 25dc6ea19c..edc3ea9149 100644 --- a/backend/windmill-object-store/src/lib.rs +++ b/backend/windmill-object-store/src/lib.rs @@ -118,6 +118,10 @@ pub fn object_store_error_to_error(err: object_store::Error) -> error::Error { pub struct ExpirableObjectStore { pub store: Arc, pub refresh: Option, + /// What locates the store's objects ([`object_store_location`]), for a store built from + /// settings. Kept with the store rather than read off the settings again, so a server + /// whose reload is still pending never names one store by another's location. + pub location: Option, } #[cfg(feature = "parquet")] @@ -155,7 +159,7 @@ impl ObjectStoreRefresh { #[cfg(feature = "parquet")] impl From> for ExpirableObjectStore { fn from(store: Arc) -> Self { - Self { store, refresh: None } + Self { store, refresh: None, location: None } } } @@ -197,6 +201,15 @@ static CACHE_OVERRIDE_GENERATION: std::sync::atomic::AtomicU64 = async fn resolve_object_store( settings_lock: &RwLock>, ) -> Option> { + resolve_object_store_with_location(settings_lock) + .await + .map(|(store, _)| store) +} + +#[cfg(feature = "parquet")] +async fn resolve_object_store_with_location( + settings_lock: &RwLock>, +) -> Option<(Arc, Option)> { let settings = settings_lock.read().await; let Some(s) = settings.as_ref() else { return None; @@ -212,18 +225,18 @@ async fn resolve_object_store( // A reload may have installed a different store while the credentials were // being minted; that one reflects newer config, so the refresh is stale. Some(current) if !Arc::ptr_eq(¤t.store, &refreshed_from) => { - Some(current.store.clone()) + Some((current.store.clone(), current.location.clone())) } Some(_) => { - let arc = new_store.store.clone(); + let found = (new_store.store.clone(), new_store.location.clone()); *settings = Some(new_store); - Some(arc) + Some(found) } // Cleared while refreshing. None => None, } } - _ => Some(s.store.clone()), + _ => Some((s.store.clone(), s.location.clone())), } } @@ -232,6 +245,13 @@ pub async fn get_object_store() -> Option> { resolve_object_store(&OBJECT_STORE_SETTINGS).await } +/// The instance object store with what locates its objects ([`object_store_location`]), +/// read together; the location is `None` for a store installed without settings. +#[cfg(feature = "parquet")] +pub async fn get_object_store_with_location() -> Option<(Arc, Option)> { + resolve_object_store_with_location(&OBJECT_STORE_SETTINGS).await +} + /// The store the dependency cache reads and writes: the worker group's override when it has one, /// the instance object store otherwise. Anything the server must also reach goes through /// [`get_object_store`] instead. @@ -422,20 +442,22 @@ pub async fn reload_object_store_setting(db: &windmill_common::DB) -> ObjectStor tracing::error!("S3 cache is not available for pro plan"); return ObjectStoreReload::Never; } - *s3_cache_settings = build_s3_client_from_settings(S3Settings { - bucket: None, - region: None, - access_key: None, - secret_key: None, - endpoint: None, - store_logs: None, - path_style: None, - allow_http: None, - port: None, - }) + *s3_cache_settings = build_object_store_from_settings( + ObjectSettings::S3(S3Settings { + bucket: None, + region: None, + access_key: None, + secret_key: None, + endpoint: None, + store_logs: None, + path_style: None, + allow_http: None, + port: None, + }), + Some(db), + ) .await .ok() - .map(|x| ExpirableObjectStore::from(x)) } else { *s3_cache_settings = None; } @@ -887,19 +909,49 @@ impl ObjectStore for FilesystemStoreIgnoringAttributes { } } +/// What locates a store's objects: endpoint, port, region and bucket (or account and +/// container, or root), never the credentials, which rotate. Two stores with the same +/// location hold the same objects. +pub fn object_store_location(resource: &ObjectStoreResource) -> String { + match resource { + ObjectStoreResource::S3(s) => format!( + "s3:{}:{}:{}:{}", + s.endpoint, + s.port.unwrap_or_default(), + s.region, + s.bucket + ), + ObjectStoreResource::Azure(a) => format!( + "azure:{}:{}:{}", + a.endpoint.as_deref().unwrap_or_default(), + a.account_name, + a.container_name + ), + ObjectStoreResource::Gcs(g) => format!("gcs:{}", g.bucket), + ObjectStoreResource::Filesystem(f) => format!("fs:{}", f.root_path), + } +} + #[cfg(feature = "parquet")] pub async fn build_object_store_from_settings( settings: ObjectSettings, init_private_key: Option<&windmill_common::DB>, ) -> error::Result { + let located = + |store: Arc, resource: ObjectStoreResource| ExpirableObjectStore { + store, + refresh: None, + location: Some(object_store_location(&resource)), + }; match settings { - ObjectSettings::S3(s3_settings) => build_s3_client_from_settings(s3_settings) - .await - .map(|x| ExpirableObjectStore::from(x)), - ObjectSettings::Azure(azure_settings) => { - let azure_blob_resource = azure_settings; - build_azure_blob_client(&azure_blob_resource).map(|x| ExpirableObjectStore::from(x)) + ObjectSettings::S3(s3_settings) => { + let s3_resource = s3_resource_from_settings(s3_settings); + build_s3_client(&s3_resource) + .await + .map(|x| located(x, ObjectStoreResource::S3(s3_resource))) } + ObjectSettings::Azure(azure_settings) => build_azure_blob_client(&azure_settings) + .map(|x| located(x, ObjectStoreResource::Azure(azure_settings))), ObjectSettings::AwsOidc(ref s3_aws_oidc_settings) => { let token_generator = crate::job_s3_helpers_oss::TokenGenerator::AsServerInstance(); let res = crate::job_s3_helpers_oss::generate_s3_aws_oidc_resource( @@ -914,17 +966,14 @@ pub async fn build_object_store_from_settings( .map(|x| ExpirableObjectStore { store: x, refresh: Some(ObjectStoreRefresh::new(settings.clone(), res.expiration())), + location: Some(object_store_location(&res)), }) } - ObjectSettings::Gcs(gcs_settings) => { - let gcs_resource = gcs_settings; - build_gcs_client(&gcs_resource) - .await - .map(|x| ExpirableObjectStore::from(x)) - } - ObjectSettings::Filesystem(fs) => { - build_filesystem_client(&fs.root_path).map(|x| ExpirableObjectStore::from(x)) - } + ObjectSettings::Gcs(gcs_settings) => build_gcs_client(&gcs_settings) + .await + .map(|x| located(x, ObjectStoreResource::Gcs(gcs_settings))), + ObjectSettings::Filesystem(fs) => build_filesystem_client(&fs.root_path) + .map(|x| located(x, ObjectStoreResource::Filesystem(fs))), } } @@ -937,14 +986,14 @@ fn none_if_empty(s: Option) -> Option { } } +/// The S3 resource instance settings resolve to, the environment filling in what they +/// leave out. #[cfg(feature = "parquet")] -pub async fn build_s3_client_from_settings( - settings: S3Settings, -) -> error::Result> { +fn s3_resource_from_settings(settings: S3Settings) -> S3Resource { let region = none_if_empty(settings.region) .unwrap_or_else(|| std::env::var("AWS_REGION").unwrap_or_else(|_| "us-east-1".to_string())); - let s3_resource = S3Resource { + S3Resource { endpoint: none_if_empty(settings.endpoint).unwrap_or_else(|| { std::env::var("S3_ENDPOINT").unwrap_or_else(|_| format!("s3.{region}.amazonaws.com")) }), @@ -959,9 +1008,7 @@ pub async fn build_s3_client_from_settings( port: settings.port, token: None, expiration: None, - }; - - build_s3_client(&s3_resource).await + } } // Resolving the default chain goes over the network (ECS/IMDS) on instances relying on an @@ -2624,6 +2671,36 @@ mod tests { reload_cache_object_store_override(&db, None).await; } + /// A store built from settings is located by where its objects are, not by how the + /// client describes itself: an S3 client prints only its bucket, so the same bucket name + /// on another endpoint would otherwise pass for the same store. + #[cfg(feature = "parquet")] + #[tokio::test] + async fn test_settings_store_location_tells_endpoints_apart() { + let s3 = |endpoint: &str| { + ObjectSettings::S3(S3Settings { + bucket: Some("windmill".to_string()), + region: Some("us-east-1".to_string()), + access_key: Some("key".to_string()), + secret_key: Some("secret".to_string()), + endpoint: Some(endpoint.to_string()), + allow_http: Some(true), + path_style: Some(true), + store_logs: None, + port: None, + }) + }; + let a = build_object_store_from_settings(s3("minio.internal:9000"), None) + .await + .unwrap(); + let b = build_object_store_from_settings(s3("s3.us-east-1.amazonaws.com"), None) + .await + .unwrap(); + assert_eq!(a.store.to_string(), b.store.to_string()); + assert!(a.location.is_some()); + assert_ne!(a.location, b.location); + } + // --- get_logs_from_store test --- #[cfg(feature = "parquet")] diff --git a/docs/ai-session-backups.md b/docs/ai-session-backups.md index 0b4c550456..3e0c747b8a 100644 --- a/docs/ai-session-backups.md +++ b/docs/ai-session-backups.md @@ -129,11 +129,58 @@ against it for the same reason. The feature is on wherever the workspace has primary storage, and off with `ai_config.sessions_storage_disabled` (the `copilot_disabled` pattern: no migration, carried by settings export and the CLI). A build without `parquet` has no routes (404), a workspace without -storage answers `enabled: false`; either turns the backup off for ten minutes, after which the +storage and nothing to stand in for it answers `enabled: false`; either turns the backup off +for ten minutes, after which the page asks again on its own (a flush for whatever is pending, and a restore), and the AI settings page tells the mirror at once when the switch is saved there (the off state is forgotten, the rows that went stale are marked again, a restore runs). +## The instance store standing in + +A workspace without storage of its own keeps its backups in the instance object store +(`object_store_cache_config`, loaded the way every other use of it is, so never with +`DISABLE_S3_STORE`; the plan is checked on every request and Pro never falls back, since a +store loaded before a switch to Pro stays loaded), under the same layout and the same +per-user key, +while the instance setting `ai_sessions_instance_storage_fallback` allows it (on unless set +to false; the instance settings page shows it under Object Storage). A build without +`private` has neither workspace storage nor the quota below, and never falls back. Every +answer says which kind of store it came from (`fallback`), and the instance store is named +(`storage_id`) by what locates its objects, the endpoint, region and bucket its settings +resolve to, in a namespace of its own: moving the instance store to another endpoint under +the same bucket name is a storage switch for the browsers, and a workspace bucket is never +taken for it. The location is kept with the loaded store, so a server whose reload is still +pending names the store it writes to. A route decides between the workspace's storage and +the instance store from the row it reads the generation from, so a push lands in the +instance store only under a generation read while the workspace had no storage. The store a +workspace's backups live in is resolved in one place +(`ai_session_backups::workspace_store`), for the routes and for the rotation's deletion of +older generations alike. + +Configuring a storage for a workspace that had none (`edit_large_file_storage_config`) bumps +the backup generation in the transaction that sets it, so everything the workspace left in +any instance store sits under a generation the routes never read again: a later return to +the instance store, whichever it is by then, starts from a newer one. Every storage settings +change then deletes from the instance store, off the request, the workspace's generations +older than the one it committed, whether the setting is on or off (copies from when it was +on may be there). Nothing live is older, whatever happens next: a deletion that is slow, cut +short, or overtaken by the storage being dropped or pointed at the instance store's own +bucket touches only generations nothing reads. Dropping the storage bumps nothing and is a +switch like any other: the rows go stale, the sessions are pushed whole into the instance +store, and the old bucket keeps its copy. On the browser side a removal owed to an instance +store (the row names it apart, `storageName` in `sessionMirrorPlan.ts`) is retired by any +answer from the workspace's own storage, since that storage being there means the +generation moved past the copy; one owed to a workspace storage still waits for that +storage, whatever the instance store answered. Copies a deletion missed stay in the +operator's bucket unread, as a deleted workspace's copies do. + +On CE the bytes in the instance store count toward the workspace's storage quota under a +storage name of their own (`_ai_sessions_fallback_`, listed by the periodic recount while +the workspace has no storage of its own, and left out when there are none), so a member +cannot fill the operator's bucket past what the workspace may use; on EE, where workspace +storage has no quota either, nothing bounds them but the per-push caps and the instance +setting. + ## Conflicts and deletion Last write wins across devices. The head carries no manifest; `pull` lists the session's prefix @@ -172,7 +219,8 @@ session whose pieces could not be written: recording it would let the next flush half-empty local state over the backup. Every answer names the storage it came from (`storage_id`, a hash of what locates the objects, -endpoint, region and bucket, not the credentials, which rotate) and the backup generation a +endpoint, region and bucket, not the credentials, which rotate; the instance store standing +in for a workspace without one is named apart, see above) and the backup generation a key rotation bumps (`backup_generation`). A sync row records both, and a row naming another storage or generation goes stale and its session is marked again: a workspace pointed at a new bucket, or whose key was rotated, holds nothing, and the server looks nowhere else, so diff --git a/frontend/src/lib/components/InstanceSettings.svelte b/frontend/src/lib/components/InstanceSettings.svelte index f51e459f77..f38c0d3892 100644 --- a/frontend/src/lib/components/InstanceSettings.svelte +++ b/frontend/src/lib/components/InstanceSettings.svelte @@ -134,11 +134,15 @@ } applyFormDefaults(nvalues) - // Apply select/select_python defaults so initialValues matches what InstanceSetting's $effect does + // Apply declared defaults before snapshotting initialValues, so a default shows without + // marking the form dirty: a select's mirrors InstanceSetting's $effect, a boolean's is + // what its toggle shows while the key is unset. for (const category of settingsKeys) { for (const s of settings[category]) { if ( - (s.fieldType === 'select' || s.fieldType === 'select_python') && + (s.fieldType === 'select' || + s.fieldType === 'select_python' || + s.fieldType === 'boolean') && nvalues[s.key] == undefined && s.defaultValue ) { diff --git a/frontend/src/lib/components/instanceSettings.ts b/frontend/src/lib/components/instanceSettings.ts index 0677648978..34079eae51 100644 --- a/frontend/src/lib/components/instanceSettings.ts +++ b/frontend/src/lib/components/instanceSettings.ts @@ -563,6 +563,17 @@ export const settings: Record = { storage: 'setting', ee_only: '' }, + { + label: 'Back AI sessions up to the instance object storage', + description: + "Browsers back their AI sessions up to their workspace's object storage, encrypted with the workspace key. When this is on and instance object storage is configured, a workspace without object storage of its own uses the instance object storage instead, under the same encryption; configuring a storage for the workspace moves its backups there and deletes what it kept in the instance storage. On by default; turn off to keep the AI sessions of such workspaces in the browser only.", + key: 'ai_sessions_instance_storage_fallback', + fieldType: 'boolean', + defaultValue: () => true, + storage: 'setting', + ee_only: '', + hideInQuickSetup: true + }, { label: 'Store audit logs in object storage', description: diff --git a/frontend/src/lib/components/sessions/sessionMirror.svelte.ts b/frontend/src/lib/components/sessions/sessionMirror.svelte.ts index b1ac4bcd06..a18e2febd6 100644 --- a/frontend/src/lib/components/sessions/sessionMirror.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionMirror.svelte.ts @@ -55,10 +55,12 @@ import { import { artifactsFingerprint, headSig, + isFallbackStorage, jsonBytes, operationsOf, planSessionPush, splitEntry, + storageName, type ChatSnapshot, type MirrorSyncState, type PlannedPush, @@ -682,9 +684,10 @@ async function pushWorkspace( return 'transient' } if (!res.enabled) return 'off' - out.storageId = res.storage_id + const storageId = storageName(res.storage_id, res.fallback) + out.storageId = storageId out.generation = res.backup_generation - const answered = `${res.storage_id}:${res.backup_generation}` + const answered = `${storageId}:${res.backup_generation}` const errors = new Set() for (const r of res.results) { if (r.error) { @@ -705,7 +708,7 @@ async function pushWorkspace( // settle, the session goes again whole. if (a.answered !== undefined && a.answered !== answered) failed.add(entry.id) a.answered = answered - a.storageId = res.storage_id + a.storageId = storageId a.generation = res.backup_generation if (errors.has(entry.id)) failed.add(entry.id) } @@ -718,12 +721,20 @@ async function pushWorkspace( [mark.storageId, ...(mark.alsoIn ?? [])].filter((s): s is string => s !== undefined) ) // Answered from a storage holding no copy: the copies are still where they - // were, and the mark waits for those storages to answer. - if (holding.size === 0 || res.storage_id === undefined) out.removedDone.push(mark) - else if (holding.has(res.storage_id)) { - holding.delete(res.storage_id) + // were, and the mark waits for those storages to answer. The workspace's own + // storage answering retires every instance store's share too: configuring it + // moved the generation past all the workspace left in any instance store. + if (holding.size === 0 || storageId === undefined) out.removedDone.push(mark) + else { + const before = holding.size + holding.delete(storageId) + if (!res.fallback) { + for (const name of [...holding]) if (isFallbackStorage(name)) holding.delete(name) + } if (holding.size === 0) out.removedDone.push(mark) - else out.removedFrom.push({ id, key: mark.key, remaining: [...holding] }) + else if (holding.size < before) { + out.removedFrom.push({ id, key: mark.key, remaining: [...holding] }) + } } } } @@ -1204,7 +1215,7 @@ async function listWorkspace(ws: string, email: string): Promise { expect((await __syncRowsForTesting(EMAIL)).some((r) => r.id === 'sr3')).toBe(false) }) + it("retires a removal owed to any instance store once the workspace's own storage answered", async () => { + const s: Session = { id: 'fb1', name: 'session-1', createdAt: 1, workspace_id: 'ws' } + sessionState.sessions = [s] + await putSession(s) + pushMock.mockResolvedValueOnce({ + enabled: true, + storage_id: 'I1', + fallback: true, + results: [{ id: 'fb1' }] + }) + await __flushForTesting() + // The operator moved the instance store: the session goes whole to the new one, and + // the row remembers the copy the old one keeps. + await putSession({ ...s, summary: 'changed' }) + pushMock.mockResolvedValue({ + enabled: true, + storage_id: 'I2', + fallback: true, + results: [{ id: 'fb1' }] + }) + await __flushForTesting() + await __flushForTesting() + const row = (await __syncRowsForTesting(EMAIL)).find((r) => r.id === 'fb1') + expect(row?.storageId).toBe('instance:I2') + expect(row?.alsoIn).toEqual(['instance:I1']) + + // The workspace got a storage of its own meanwhile, which moved the generation past + // everything it left in either instance store: that storage's answer settles the + // removal. + deleteSession('fb1') + await flush() + pushMock.mockResolvedValue({ enabled: true, storage_id: 'A', results: [{ id: 'fb1' }] }) + await __flushForTesting() + expect(pushMock.mock.lastCall?.[0].requestBody.removed).toEqual(['fb1']) + expect(removalKeys()).toEqual([]) + expect((await __syncRowsForTesting(EMAIL)).some((r) => r.id === 'fb1')).toBe(false) + }) + + it('waits for the storage a workspace dropped, whatever the instance store answered', async () => { + const s: Session = { id: 'fb2', name: 'session-1', createdAt: 1, workspace_id: 'ws' } + sessionState.sessions = [s] + await putSession(s) + pushMock.mockResolvedValueOnce({ enabled: true, storage_id: 'A', results: [{ id: 'fb2' }] }) + await __flushForTesting() + // The workspace dropped its storage: the session goes whole to the instance store, + // and the row remembers the copy A keeps. + await putSession({ ...s, summary: 'changed' }) + pushMock.mockResolvedValue({ + enabled: true, + storage_id: 'I', + fallback: true, + results: [{ id: 'fb2' }] + }) + await __flushForTesting() + await __flushForTesting() + const row = (await __syncRowsForTesting(EMAIL)).find((r) => r.id === 'fb2') + expect(row?.storageId).toBe('instance:I') + expect(row?.alsoIn).toEqual(['A']) + + // Deleted while on the instance store: its copy goes, and the mark waits for A. + deleteSession('fb2') + await flush() + await __flushForTesting() + expect(removalKeys()).toEqual(['r::fb2::ws']) + expect((await __syncRowsForTesting(EMAIL)).find((r) => r.id === 'fb2')?.storageId).toBe('A') + pushMock.mockResolvedValue({ enabled: true, storage_id: 'A', results: [{ id: 'fb2' }] }) + await __flushForTesting() + expect(pushMock.mock.lastCall?.[0].requestBody.removed).toEqual(['fb2']) + expect(removalKeys()).toEqual([]) + }) + it("removes a moved session's old copy from the storage that held it, whatever its old workspace is on now", async () => { const s: Session = { id: 'mv2', name: 'session-1', createdAt: 1, workspace_id: 'ws' } sessionState.sessions = [s] diff --git a/frontend/src/lib/components/sessions/sessionMirrorPlan.ts b/frontend/src/lib/components/sessions/sessionMirrorPlan.ts index c61bcf2763..a664e61f5e 100644 --- a/frontend/src/lib/components/sessions/sessionMirrorPlan.ts +++ b/frontend/src/lib/components/sessions/sessionMirrorPlan.ts @@ -31,9 +31,9 @@ export interface MirrorSyncState { /** The user deleted the session and its removal mark could not be written to * localStorage (full): the row itself carries the removal, until it lands. */ removed?: boolean - /** The storage the push landed in, as the server names it, and the backup generation - * (bumped by a workspace key rotation) it landed under. A row recorded against another - * storage or generation describes objects the server no longer looks at. */ + /** The storage the push landed in (`storageName`), and the backup generation (bumped by + * a workspace key rotation) it landed under. A row recorded against another storage or + * generation describes objects the server no longer looks at. */ storageId?: string generation?: number /** Other storages this workspace was on that still hold a copy of the backup (a switch @@ -48,6 +48,34 @@ export interface MirrorSyncState { staging?: { chats: string[]; images: string[]; items: string[]; versions: string[] } } +const FALLBACK_STORAGE_PREFIX = 'instance:' + +/** + * How a storage the server answered from is named in the sync rows and the removal marks: + * by the id the server gives it, the instance object store standing in for a workspace + * without storage of its own (`fallback` on the answer) told apart from a workspace's own. + * A removal owed to an instance store is retired by any answer from the workspace's own + * storage (configuring one moves the backup generation past everything the workspace left + * in any instance store, so none of it is read again), where one owed to a workspace + * storage waits for that storage. + */ +export function storageName(id: string, fallback: boolean | undefined): string +export function storageName( + id: string | undefined, + fallback: boolean | undefined +): string | undefined +export function storageName( + id: string | undefined, + fallback: boolean | undefined +): string | undefined { + if (id === undefined) return undefined + return fallback ? FALLBACK_STORAGE_PREFIX + id : id +} + +export function isFallbackStorage(name: string): boolean { + return name.startsWith(FALLBACK_STORAGE_PREFIX) +} + /** * The part of a session record the backup keeps. Left out on purpose: `name` (a * per-browser counter), the unsent-draft fields (`pending_*`, `draftPrompt`, diff --git a/frontend/src/lib/components/workspaceSettings/AISettings.svelte b/frontend/src/lib/components/workspaceSettings/AISettings.svelte index 7e8b9cc76e..5bc7974af2 100644 --- a/frontend/src/lib/components/workspaceSettings/AISettings.svelte +++ b/frontend/src/lib/components/workspaceSettings/AISettings.svelte @@ -661,7 +661,7 @@