diff --git a/backend/.sqlx/query-8eee14066c86b4a4ef921576277c8a287bf1ff4b9d4301e3ea0efd8077936aff.json b/backend/.sqlx/query-8eee14066c86b4a4ef921576277c8a287bf1ff4b9d4301e3ea0efd8077936aff.json new file mode 100644 index 0000000000..552a38a0d4 --- /dev/null +++ b/backend/.sqlx/query-8eee14066c86b4a4ef921576277c8a287bf1ff4b9d4301e3ea0efd8077936aff.json @@ -0,0 +1,30 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT workspace_settings.workspace_id AS \"id!\",\n workspace_settings.ai_config->'sessions_retention_days' AS retention\n FROM workspace_settings\n LEFT JOIN usr ON usr.workspace_id = workspace_settings.workspace_id AND usr.email = $2\n WHERE workspace_settings.workspace_id = ANY($1)\n AND ($3 OR (usr.email IS NOT NULL AND NOT usr.disabled))", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id!", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "retention", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "TextArray", + "Text", + "Bool" + ] + }, + "nullable": [ + false, + null + ] + }, + "hash": "8eee14066c86b4a4ef921576277c8a287bf1ff4b9d4301e3ea0efd8077936aff" +} 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 aa0b22ac22..51d209f654 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -05c7050da4caedaf8e19a597960beb5d5f5254a9 +5f68c8c351ffc92feccffe69a857b60be376464e 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 f76385d4e1..6b33a4141e 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() @@ -87,7 +135,8 @@ async fn rotate(base: &str, key: &str) -> anyhow::Result<()> { } /// The user's prefix on disk, `windmill_ai_sessions/{w_id}/g{generation}/{email hash}`, -/// under whichever generation is current. +/// under the newest generation: deleting an older generation's objects leaves its +/// directories behind, and `read_dir` order differs across filesystems. fn user_root(storage_dir: &std::path::Path, email: &str) -> std::path::PathBuf { let workspace = storage_dir.join("windmill_ai_sessions/test-workspace"); let hash = calculate_hash(email); @@ -96,8 +145,18 @@ fn user_root(storage_dir: &std::path::Path, email: &str) -> std::path::PathBuf { .into_iter() .flatten() .flatten() - .map(|entry| entry.path().join(&hash)) - .find(|path| path.exists()) + .filter_map(|entry| { + let generation: i64 = entry + .file_name() + .to_str()? + .strip_prefix('g')? + .parse() + .ok()?; + Some((generation, entry.path().join(&hash))) + }) + .filter(|(_, path)| path.exists()) + .max_by_key(|(generation, _)| *generation) + .map(|(_, path)| path) .expect("the user has backups under the current key") } @@ -132,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!([])); @@ -945,6 +1006,96 @@ async fn test_backups_round_trip_encrypted_and_scoped_to_the_user( .await?; assert_eq!(resp.status(), 200); + // An incremental part changing more than one object unlists the session before its + // writes, so one write failing after another landed leaves it absent rather than listed + // as a mix of old and new pieces. A directory planted at `artifacts.json` fails that write. + let s10_head = + json!({ "id": "s10", "workspace_id": "test-workspace", "createdAt": 10, "chatId": "c1" }); + let resp = push( + &base, + "SECRET_TOKEN", + json!({ + "owner": "test@windmill.dev", + "sessions": [{ "id": "s10", "whole": true, "head": s10_head, "chats": s9_chats(&["c1"]), "artifacts": { "items": ["a1"] } }] + }), + ) + .await?; + assert_eq!(resp.status(), 200); + let s10_dir = user_root(storage_dir.path(), "test@windmill.dev").join("sessions/s10"); + let artifacts_path = s10_dir.join("artifacts.json"); + std::fs::remove_file(&artifacts_path)?; + std::fs::create_dir(&artifacts_path)?; + let resp = push( + &base, + "SECRET_TOKEN", + json!({ + "owner": "test@windmill.dev", + "sessions": [{ "id": "s10", "chats": s9_chats(&["c2"]), "artifacts": { "items": ["a2"] } }] + }), + ) + .await?; + assert_eq!(resp.status(), 200); + let answer: Value = resp.json().await?; + assert!( + answer["results"][0]["error"].is_string(), + "the artifacts write must fail: {answer}" + ); + assert!(answer["results"][0]["needs_whole"].is_null()); + assert!( + s10_dir.join("chats/c2.json").is_file(), + "the chat landed before the artifacts failed" + ); + let s10_listed = |listing: Value| { + listing["sessions"] + .as_array() + .unwrap() + .iter() + .any(|s| s["id"] == "s10") + }; + assert!(!s10_listed(list(&base, "SECRET_TOKEN").await?)); + assert_eq!( + pull(&base, "SECRET_TOKEN", &["s10"]).await?["sessions"], + json!([]) + ); + let resp = push( + &base, + "SECRET_TOKEN", + json!({ + "owner": "test@windmill.dev", + "sessions": [{ "id": "s10", "chats": s9_chats(&["c3"]) }] + }), + ) + .await?; + assert_eq!(resp.status(), 200); + let answer: Value = resp.json().await?; + assert_eq!(answer["results"][0]["needs_whole"], true); + assert!(!s10_listed(list(&base, "SECRET_TOKEN").await?)); + std::fs::remove_dir(&artifacts_path)?; + let resp = push( + &base, + "SECRET_TOKEN", + json!({ + "owner": "test@windmill.dev", + "sessions": [{ "id": "s10", "whole": true, "head": s10_head, "chats": s9_chats(&["c1", "c2", "c3"]), "artifacts": { "items": ["a2"] } }] + }), + ) + .await?; + assert_eq!(resp.status(), 200); + assert!(s10_listed(list(&base, "SECRET_TOKEN").await?)); + let pulled = pull(&base, "SECRET_TOKEN", &["s10"]).await?; + assert_eq!( + pulled["sessions"][0]["artifacts"], + json!({ "items": ["a2"] }) + ); + assert_eq!(pulled_chats(pulled), vec!["c1", "c2", "c3"]); + let resp = push( + &base, + "SECRET_TOKEN", + json!({ "owner": "test@windmill.dev", "removed": ["s10"] }), + ) + .await?; + assert_eq!(resp.status(), 200); + // Removal empties both prefixes. let resp = push( &base, @@ -1083,3 +1234,349 @@ async fn test_backup_writes_are_refused_for_the_wrong_owner_token_or_id( assert!(files_under(storage_dir.path()).is_empty()); 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 { + 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); + + // 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; + 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/tests/bun_jobs.rs b/backend/tests/bun_jobs.rs index 590963b291..63d5341ba3 100644 --- a/backend/tests/bun_jobs.rs +++ b/backend/tests/bun_jobs.rs @@ -1209,6 +1209,100 @@ export function main() { Ok(()) } +/// A deployed flow runs an inline step as the `flow_node` its deploy rewrote it into, +/// a `FlowScript` job rather than the preview job the editor runs. A workflow-as-code +/// step's `task()` children must dispatch from that kind too, as re-runs of the same +/// node, or the step passes its editor test and fails once deployed. +/// +/// The step is cached: a child that shared the parent's result-cache key would hand +/// its own result (`10`) back to the parent on resume, in place of the workflow's. +#[sqlx::test(fixtures("base", "wac_flow_script"))] +async fn test_bun_wac_task_dispatch_from_flow_script(db: Pool) -> anyhow::Result<()> { + use windmill_common::flows::FlowNodeId; + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let node = FlowNodeId(3000000000000011); + let job = RunJob::from(JobPayload::FlowScript { + id: node, + path: "f/system/wac_flow_script/a".to_string(), + language: ScriptLang::Bun, + cache_ttl: Some(60), + cache_ignore_s3_path: None, + dedicated_worker: None, + concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(), + }) + .arg("n", serde_json::json!(5)) + .run_until_complete(&db, false, port) + .await; + + assert_eq!( + job.json_result().unwrap(), + serde_json::json!({"doubled": 10}) + ); + + let children: Vec<(String, Option, Option)> = sqlx::query_as( + "SELECT kind::text, runnable_id, cache_ttl FROM v2_job WHERE parent_job = $1", + ) + .bind(job.id) + .fetch_all(&db) + .await?; + assert_eq!( + children, + vec![("flowscript".to_string(), Some(node.0), None)], + "the task child re-runs the parent's flow node, outside the result cache" + ); + Ok(()) +} + +/// `task(fn, { cache_ttl })` on an inline task of a deployed flow's step: the child runs +/// the parent's code with the parent's arguments, so its result-cache key carries its +/// step key, or the parent and every sibling would read its result back as their own. +#[sqlx::test(fixtures("base", "wac_flow_script"))] +async fn test_bun_wac_inline_task_cache_is_per_task(db: Pool) -> anyhow::Result<()> { + use windmill_common::flows::FlowNodeId; + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let payload = || JobPayload::FlowScript { + id: FlowNodeId(3000000000000012), + path: "f/system/wac_flow_script/a".to_string(), + language: ScriptLang::Bun, + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: None, + concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(), + }; + + let mut children_from_cache = Vec::new(); + for _ in 0..2 { + let job = RunJob::from(payload()) + .arg("n", serde_json::json!(5)) + .run_until_complete(&db, false, port) + .await; + assert_eq!( + job.json_result().unwrap(), + serde_json::json!({"doubled": 10, "tripled": 15}) + ); + let from_cache: i64 = sqlx::query_scalar( + "SELECT count(*) FROM job_logs l JOIN v2_job j ON j.id = l.job_id \ + WHERE j.parent_job = $1 AND l.logs LIKE '%found in cache%'", + ) + .bind(job.id) + .fetch_one(&db) + .await?; + children_from_cache.push(from_cache); + } + assert_eq!( + children_from_cache, + vec![0, 2], + "the second run serves each task from its own cache entry" + ); + Ok(()) +} + // ============================================================================ // Environment Variable Tests // ============================================================================ diff --git a/backend/tests/fixtures/wac_flow_script.sql b/backend/tests/fixtures/wac_flow_script.sql new file mode 100644 index 0000000000..1b780c6cf4 --- /dev/null +++ b/backend/tests/fixtures/wac_flow_script.sql @@ -0,0 +1,53 @@ +-- A deployed flow whose inline bun step is workflow-as-code calling task(), in the +-- shape the deploy leaves behind: the RawScript module rewritten into a flow_node that +-- the step then runs as a FlowScript job. No lock, so the worker resolves +-- windmill-client at run time like the other bun fixtures. +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', '', '', +'f/system/wac_flow_script', +'{}', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"n":{"type":"integer","description":""}},"required":[],"type":"object"}', +'{"modules":[{"id":"a","value":{"type":"flowscript","id":3000000000000011,"language":"bun","input_transforms":{"n":{"expr":"flow_input.n","type":"javascript"}}}}]}', +'system' +); + +INSERT INTO public.flow_node(id, workspace_id, path, hash_v2, lock, code) VALUES ( +3000000000000011, +'test-workspace', +'f/system/wac_flow_script', +'0000000000000000000000000000000000000000000000000000000000000011', +NULL, +E'import { workflow, task } from "windmill-client"; + +const double = task(async (n: number) => { + return n * 2; +}); + +export const main = workflow(async (n: number) => { + const d = await double(n); + return { doubled: d }; +});' +); + +-- The same flow's step with two tasks that cache their own result. +INSERT INTO public.flow_node(id, workspace_id, path, hash_v2, lock, code) VALUES ( +3000000000000012, +'test-workspace', +'f/system/wac_flow_script', +'0000000000000000000000000000000000000000000000000000000000000012', +NULL, +E'import { workflow, task } from "windmill-client"; + +const double = task(async (n: number) => { + return n * 2; +}, { cache_ttl: 60 }); +const triple = task(async (n: number) => { + return n * 3; +}, { cache_ttl: 60 }); + +export const main = workflow(async (n: number) => { + const d = await double(n); + const t = await triple(n); + return { doubled: d, tripled: t }; +});' +); diff --git a/backend/tests/session_workspace_status.rs b/backend/tests/session_workspace_status.rs index b4fc98da5e..2ed4ac503a 100644 --- a/backend/tests/session_workspace_status.rs +++ b/backend/tests/session_workspace_status.rs @@ -3,18 +3,23 @@ //! extractor actually grants. Membership is not the only path: a superadmin is authed into //! any existing workspace without a `usr` row, and `admins` has no `usr` rows at all, so //! answering from `usr` alone reports live workspaces as unresolvable and the client deletes -//! sessions that still work. +//! sessions that still work. `POST /workspaces/session_workspace_retention`, the AI session +//! retention the same client deletes its own copies by, is a workspace setting and answers to +//! the stricter bar, which is why the two are separate routes and tested together. use serde_json::json; use sqlx::{Pool, Postgres}; use std::collections::HashMap; use windmill_test_utils::*; -async fn status(port: u16, token: &str, ids: &[&str]) -> anyhow::Result> { +async fn post( + port: u16, + route: &str, + token: &str, + ids: &[&str], +) -> anyhow::Result { let resp = reqwest::Client::new() - .post(format!( - "http://localhost:{port}/api/workspaces/session_workspace_status" - )) + .post(format!("http://localhost:{port}/api/workspaces/{route}")) .header("Authorization", format!("Bearer {token}")) .json(&json!({ "workspace_ids": ids })) .send() @@ -23,6 +28,14 @@ async fn status(port: u16, token: &str, ids: &[&str]) -> anyhow::Result anyhow::Result> { + post(port, "session_workspace_status", token, ids).await +} + +async fn retention(port: u16, token: &str, ids: &[&str]) -> anyhow::Result> { + post(port, "session_workspace_retention", token, ids).await +} + #[sqlx::test(fixtures("base", "session_workspace_status"))] async fn test_superadmin_reaches_workspaces_without_a_usr_row( db: Pool, @@ -60,3 +73,48 @@ async fn test_superadmin_reaches_workspaces_without_a_usr_row( Ok(()) } + +/// The retention a browser deletes its own copies by is a workspace setting, so unlike the +/// status it is told only to a caller the authed extractor would let in. +#[sqlx::test(fixtures("base", "session_workspace_status"))] +async fn test_session_retention_is_told_only_to_members_who_can_be_authed( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let ids = ["foreign-workspace", "test-workspace", "no-such-workspace"]; + sqlx::query( + "UPDATE workspace_settings SET ai_config = '{\"sessions_retention_days\": 7}' \ + WHERE workspace_id IN ('test-workspace', 'foreign-workspace')", + ) + .execute(&db) + .await?; + + // test@windmill.dev is a superadmin: authed into every workspace that exists. + let sa = retention(port, "SECRET_TOKEN", &ids).await?; + assert_eq!(sa["test-workspace"], 7); + assert_eq!(sa["foreign-workspace"], 7); + assert!(!sa.contains_key("no-such-workspace")); + + // test2@windmill.dev is a member of test-workspace only. + let usr = retention(port, "SECRET_TOKEN_2", &ids).await?; + assert_eq!(usr["test-workspace"], 7); + assert!(!usr.contains_key("foreign-workspace")); + + // A disabled membership still reconciles its sessions — the status stays `active` — but + // cannot be authed into the workspace, so it is told no setting. + sqlx::query("UPDATE usr SET disabled = true WHERE workspace_id = 'test-workspace'") + .execute(&db) + .await?; + assert_eq!( + status(port, "SECRET_TOKEN_2", &ids).await?["test-workspace"], + "active" + ); + assert!(!retention(port, "SECRET_TOKEN_2", &ids) + .await? + .contains_key("test-workspace")); + + Ok(()) +} diff --git a/backend/windmill-api-jobs/src/execution.rs b/backend/windmill-api-jobs/src/execution.rs index e897bf5eaf..310e44ae1f 100644 --- a/backend/windmill-api-jobs/src/execution.rs +++ b/backend/windmill-api-jobs/src/execution.rs @@ -38,8 +38,8 @@ use windmill_common::{ FlowVersionInfo, DB, }; use windmill_queue::{ - cancel_job, get_result_and_success_by_id_from_flow, push, PushArgs, PushArgsOwned, - PushIsolationLevel, + cancel_job, get_result_and_success_by_id_from_flow, parse_result_object, push, PushArgs, + PushArgsOwned, PushIsolationLevel, }; use crate::types::RunJobQuery; @@ -374,9 +374,9 @@ pub async fn run_wait_result_internal( } pub fn result_to_response(result: Box, success: bool) -> error::Result { - let composite_result = serde_json::from_str::(result.get()); + let composite_result = parse_result_object::(result.get()); match composite_result { - Ok(WindmillCompositeResult { + Some(WindmillCompositeResult { windmill_status_code, windmill_content_type, windmill_headers, @@ -1192,4 +1192,13 @@ mod result_to_response_tests { assert!(res.is_err(), "hop-by-hop header must be rejected: {name}"); } } + + #[tokio::test] + async fn array_result_is_not_a_composite_response() { + let json = r#"[201,"text/html",null,null,"

hi

"]"#; + let resp = result_to_response(raw(json), true).expect("response"); + + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!(body_bytes(resp).await, json.as_bytes()); + } } diff --git a/backend/windmill-api-workspaces/src/ai_session_backups.rs b/backend/windmill-api-workspaces/src/ai_session_backups.rs index bbdbfbc8c9..8e7b75e1e2 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,77 @@ 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, 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 @@ -91,9 +150,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 +176,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 +218,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..3330ed697f 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -227,6 +227,10 @@ pub fn global_service() -> Router { .route("/list", get(list_workspaces)) .route("/users", get(user_workspaces)) .route("/session_workspace_status", post(session_workspace_status)) + .route( + "/session_workspace_retention", + post(session_workspace_retention), + ) .route("/create", post(create_workspace)) .route("/create_fork", post(deprecated_create_workspace_fork)) .route("/exists", post(exists_workspace)) @@ -2093,6 +2097,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 +2123,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, @@ -5603,6 +5633,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 @@ -5652,6 +5690,42 @@ async fn session_workspace_status( Ok(Json(statuses)) } +/// The AI session retention a browser deletes its local copies by (docs/ai-session-backups.md). +/// Its own route, not a field on the status above, whose shape an older tab still reads. Unlike +/// a status, it answers only for a workspace this caller can be authed into: a setting is the +/// workspace's to tell, so a disabled membership gets none though its sessions still reconcile. +async fn session_workspace_retention( + Extension(db): Extension, + authed: ApiAuthed, + Json(req): Json, +) -> JsonResult> { + if req.workspace_ids.len() > 1000 { + return Err(Error::BadRequest( + "Too many workspace ids (max 1000)".to_string(), + )); + } + let email = &authed.email; + let is_superadmin = windmill_api_auth::is_super_admin_authed(&db, &authed).await?; + let rows = sqlx::query!( + "SELECT workspace_settings.workspace_id AS \"id!\", + workspace_settings.ai_config->'sessions_retention_days' AS retention + FROM workspace_settings + LEFT JOIN usr ON usr.workspace_id = workspace_settings.workspace_id AND usr.email = $2 + WHERE workspace_settings.workspace_id = ANY($1) + AND ($3 OR (usr.email IS NOT NULL AND NOT usr.disabled))", + &req.workspace_ids[..], + email, + is_superadmin, + ) + .fetch_all(&db) + .await?; + let days = rows + .into_iter() + .filter_map(|r| sessions_retention_days(r.retention.as_ref()).map(|days| (r.id, days))) + .collect(); + Ok(Json(days)) +} + /// The instance critical alert channels belong to the instance operator, who on cloud is /// not the workspace owner and never opted into a tenant's job failures. Fork workspaces run /// throwaway copies of their parent's runnables, so instance-wide operational alerting must diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 9f742a129b..c88a298646 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -1277,6 +1277,37 @@ paths: - archived - deleted + /workspaces/session_workspace_retention: + post: + summary: get the AI session retention of workspaces referenced by client-side sessions + operationId: getSessionWorkspaceRetention + tags: + - workspace + requestBody: + required: true + content: + application/json: + schema: + type: object + properties: + workspace_ids: + type: array + items: + type: string + required: + - workspace_ids + responses: + "200": + description: >- + map of workspace id to its `ai_config.sessions_retention_days`; a workspace + without a retention, or one the caller cannot be authenticated into, is absent + content: + application/json: + schema: + type: object + additionalProperties: + type: integer + /w/{workspace}/workspaces/get_as_superadmin: get: summary: get workspace as super admin (require to be super admin) @@ -13222,6 +13253,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 +13307,8 @@ paths: type: string backup_generation: type: integer + fallback: + type: boolean sessions: type: array items: @@ -13327,6 +13363,8 @@ paths: type: string backup_generation: type: integer + fallback: + type: boolean results: type: array items: @@ -28441,6 +28479,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 fca9c35fca..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::{ - 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,12 @@ 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, } impl Backend { @@ -108,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 { @@ -118,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 { @@ -307,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(), @@ -408,34 +474,61 @@ 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", - ) - .bind(w_id) - .fetch_optional(db) - .await? - .unwrap_or((None, 0)); + 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 (_, resource) = - crate::job_helpers_oss::get_workspace_s3_resource(authed, db, None, w_id, None).await?; - let Some(resource) = resource else { - 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`). + 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, + retention_days, + fallback, + })) } #[derive(Serialize)] @@ -455,6 +548,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")] @@ -462,7 +561,9 @@ struct ListResponse { } /// A session is listed once a push entry of it landed whole (its marker is written last); -/// a push that failed before that left objects the listing does not name. +/// a push that failed before that left objects the listing does not name. One past the +/// workspace's retention is not listed either, whether or not the sweep has reached it, so a +/// browser never restores it. async fn list( authed: ApiAuthed, Extension(db): Extension, @@ -474,11 +575,13 @@ async fn list( enabled: false, storage_id: None, backup_generation: None, + fallback: false, sessions: vec![], truncated: false, })); }; 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. @@ -494,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, @@ -526,6 +622,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 +691,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 +938,7 @@ async fn pull( enabled: false, storage_id: None, backup_generation: None, + fallback: false, sessions: vec![], deferred: vec![], })); @@ -861,6 +961,7 @@ async fn pull( enabled: true, storage_id: Some(backend.storage_id), backup_generation: Some(backend.generation), + fallback: backend.fallback, sessions, deferred, })) @@ -952,6 +1053,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, } @@ -1167,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 @@ -1222,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, @@ -1241,6 +1557,7 @@ async fn push( enabled: false, storage_id: None, backup_generation: None, + fallback: false, results: vec![], })); }; @@ -1294,16 +1611,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 +1628,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-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/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/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index c4058dba6f..ab91d54287 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -654,6 +654,16 @@ pub struct ResultMetadata { pub wm_failure: Option, } +/// Parses a marker struct out of a job result, which only an object can carry. +/// A derived `Deserialize` also accepts an array, filling fields by position, so +/// without the check a result like `[[], "boom"]` reads as `wm_failure: "boom"`. +pub fn parse_result_object(result: &str) -> Option { + if !result.trim_start().starts_with('{') { + return None; + } + serde_json::from_str(result).ok() +} + /// Sentinel `error.name` we inject into a result when retagging a successful /// run as a failure due to `wm_failure`. Used downstream to detect that /// the result is already in the standard `{ error: { name, message }, ... }` @@ -674,8 +684,7 @@ pub fn is_pre_shaped_wm_failure_result(result: &str) -> bool { struct NameOnly { name: String, } - serde_json::from_str::(result) - .ok() + parse_result_object::(result) .and_then(|m| m.error) .map(|e| e.name == MANUAL_FAILURE_ERROR_NAME) .unwrap_or(false) @@ -721,7 +730,7 @@ impl ValidableJson for Box { } fn result_metadata(&self) -> ResultMetadata { - serde_json::from_str::(self.get()).unwrap_or_default() + parse_result_object::(self.get()).unwrap_or_default() } fn size(&self) -> usize { @@ -774,6 +783,10 @@ impl ValidableJson for serde_json::Value { } fn result_metadata(&self) -> ResultMetadata { + // An array would decode positionally, see `parse_result_object`. + if !self.is_object() { + return ResultMetadata::default(); + } serde_json::from_value::(self.clone()).unwrap_or_default() } @@ -7876,3 +7889,36 @@ mod git_sync_concurrency_key_tests { assert!(a.len() <= 255 && b.len() <= 255); } } + +#[cfg(test)] +mod result_metadata_tests { + use super::{ResultMetadata, ValidableJson}; + use serde_json::value::RawValue; + + fn from_raw(json: &str) -> ResultMetadata { + RawValue::from_string(json.to_string()) + .unwrap() + .result_metadata() + } + + fn from_value(json: &str) -> ResultMetadata { + serde_json::from_str::(json) + .unwrap() + .result_metadata() + } + + #[test] + fn array_result_carries_no_markers() { + for json in [r#"[["label"], "boom"]"#, r#"[null, "boom"]"#] { + for meta in [from_raw(json), from_value(json)] { + assert!( + meta.wm_labels.is_none() && meta.wm_failure.is_none(), + "{json}" + ); + } + } + let meta = from_raw(r#"{"wm_labels": ["label"], "wm_failure": "boom"}"#); + assert_eq!(meta.wm_labels, Some(vec!["label".to_string()])); + assert_eq!(meta.wm_failure.as_deref(), Some("boom")); + } +} diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 7d45cab178..117fbda086 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -2874,15 +2874,17 @@ pub async fn handle_wac_v2_output( .collect(); // Resolve job_payload once (same for all children since they re-run - // the parent script) + // the parent script). The step's cache setting is for the workflow's + // result; a task is cached only through its own `cache_ttl` option, + // under a key of its own (see `cached_result_path`). let job_payload_template = match job.kind { JobKind::Script => { if let Some(hash) = job.runnable_id { Ok(JobPayload::ScriptHash { hash, path: job.runnable_path.clone().unwrap_or_default(), - cache_ttl: job.cache_ttl, - cache_ignore_s3_path: job.cache_ignore_s3_path, + cache_ttl: None, + cache_ignore_s3_path: None, dedicated_worker: None, language: job.script_lang.unwrap_or(ScriptLang::Bun), priority: job.priority, @@ -2897,6 +2899,27 @@ pub async fn handle_wac_v2_output( )) } } + // A deployed flow runs an inline step as the `flow_node` its deploy + // rewrote it into; the child re-runs that node the way a `Script` + // child re-runs its hash, so `runnable_id` (the checkpoint's source + // hash) stays the same across parent and children. + JobKind::FlowScript => { + if let Some(id) = job.runnable_id { + Ok(JobPayload::FlowScript { + id: windmill_common::flows::FlowNodeId(id.0), + path: job.runnable_path.clone().unwrap_or_default(), + language: job.script_lang.unwrap_or(ScriptLang::Bun), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: None, + concurrency_settings: ConcurrencySettings::default(), + }) + } else { + Err(error::Error::internal_err( + "WAC v2 FlowScript job missing runnable_id".to_string(), + )) + } + } JobKind::Preview => { let row: Option<(Option, Option)> = sqlx::query_as( "SELECT raw_code, raw_lock FROM v2_job WHERE id = $1 AND workspace_id = $2", @@ -2912,8 +2935,8 @@ pub async fn handle_wac_v2_output( hash: None, language: job.script_lang.unwrap_or(ScriptLang::Bun), lock: lock, - cache_ttl: job.cache_ttl, - cache_ignore_s3_path: job.cache_ignore_s3_path, + cache_ttl: None, + cache_ignore_s3_path: None, dedicated_worker: None, concurrency_settings: ConcurrencySettingsWithCustom::default(), debouncing_settings: DebouncingSettings::default(), @@ -3012,6 +3035,12 @@ pub async fn handle_wac_v2_output( let mut pushed_ids: Vec = Vec::with_capacity(num_steps); let push_result: error::Result<()> = async { for (step, (_, child_uuid)) in steps.iter().zip(job_ids.iter()) { + // A task with a runnable of its own (a deployed script or flow) queues + // at that runnable's priority; any other task is the parent's code and + // queues at the parent's. + let own_runnable = matches!(step.dispatch_type.as_str(), "script" | "flow") + && !step.script.starts_with("./"); + // Resolve job payload based on dispatch_type let (job_payload, child_args, is_external, on_behalf_of) = match step.dispatch_type.as_str() { @@ -3025,8 +3054,8 @@ pub async fn handle_wac_v2_output( hash: None, language: module.language, lock: module.lock, - cache_ttl: job.cache_ttl, - cache_ignore_s3_path: job.cache_ignore_s3_path, + cache_ttl: None, + cache_ignore_s3_path: None, dedicated_worker: None, concurrency_settings: ConcurrencySettingsWithCustom::default(), debouncing_settings: DebouncingSettings::default(), @@ -3110,7 +3139,8 @@ pub async fn handle_wac_v2_output( let mut job_payload = job_payload; if let Some(cache_ttl) = step.cache_ttl { match &mut job_payload { - JobPayload::ScriptHash { cache_ttl: ref mut ct, .. } => { + JobPayload::ScriptHash { cache_ttl: ref mut ct, .. } + | JobPayload::FlowScript { cache_ttl: ref mut ct, .. } => { *ct = Some(cache_ttl) } JobPayload::Code(ref mut code) => code.cache_ttl = Some(cache_ttl), @@ -3122,7 +3152,8 @@ pub async fn handle_wac_v2_output( || step.concurrency_time_window_s.is_some() { match &mut job_payload { - JobPayload::ScriptHash { concurrency_settings: ref mut cs, .. } => { + JobPayload::ScriptHash { concurrency_settings: ref mut cs, .. } + | JobPayload::FlowScript { concurrency_settings: ref mut cs, .. } => { if let Some(limit) = step.concurrent_limit { cs.concurrent_limit = Some(limit); } @@ -3188,13 +3219,14 @@ pub async fn handle_wac_v2_output( job.visible_to_owner, step.tag.clone().or_else(|| Some(job.tag.clone())), step.timeout.or(job.timeout), - None, // flow_step_id - step.priority, // priority_override - None, // authed - false, // running - None, // end_user_email - None, // trigger - None, // suspended_mode + None, // flow_step_id + step.priority + .or(if own_runnable { None } else { job.priority }), + None, // authed + false, // running + None, // end_user_email + None, // trigger + None, // suspended_mode ) .await?; diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 0a5deb52d2..9c012fd117 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -1559,7 +1559,7 @@ pub async fn cached_result_path( client: &AuthedClient, job: &MiniPulledJob, raw_data: Option<&RawData>, -) -> String { +) -> windmill_common::error::Result { let mut hasher = sha2::Sha256::new(); hasher.update(&[job.kind as u8]); if let Some(ScriptHash(hash)) = job.runnable_id { @@ -1574,6 +1574,13 @@ pub async fn cached_result_path( _ => {} } } + // A workflow-as-code task child runs its parent's code with the parent's + // arguments; the step it executes is what tells its result from the parent's + // and from its siblings'. + if let Some(step_key) = wac_executing_key(db, job).await? { + hasher.update(b"wac_step:"); + hasher.update(step_key.as_bytes()); + } hash_args( db, client, @@ -1584,7 +1591,26 @@ pub async fn cached_result_path( job.cache_ignore_s3_path.unwrap_or(false), ) .await; - format!("g/results/{:064x}", hasher.finalize()) + Ok(format!("g/results/{:064x}", hasher.finalize())) +} + +/// The checkpoint step key a workflow-as-code parent seeded for this child at push +/// time; `None` for any job that is not such a child. +async fn wac_executing_key( + db: &DB, + job: &MiniPulledJob, +) -> windmill_common::error::Result> { + if job.parent_job.is_none() || job.flow_step_id.is_some() { + return Ok(None); + } + let key: Option> = sqlx::query_scalar( + "SELECT workflow_as_code_status->'_checkpoint'->>'_executing_key' \ + FROM v2_job_status WHERE id = $1", + ) + .bind(job.id) + .fetch_optional(db) + .await?; + Ok(key.flatten()) } #[cfg(feature = "parquet")] diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index a269e72f46..1529d474fa 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -32,8 +32,8 @@ use windmill_common::bench::{BenchmarkInfo, BenchmarkIter}; use windmill_queue::{ append_logs, asset_dispatch, get_mini_completed_job, is_pre_shaped_wm_failure_result, - CanceledBy, FlowRunners, JobCompleted, MiniCompletedJob, MiniPulledJob, ValidableJson, - WrappedError, INIT_SCRIPT_TAG, MANUAL_FAILURE_ERROR_NAME, + parse_result_object, CanceledBy, FlowRunners, JobCompleted, MiniCompletedJob, MiniPulledJob, + ValidableJson, WrappedError, INIT_SCRIPT_TAG, MANUAL_FAILURE_ERROR_NAME, }; use serde_json::{json, value::RawValue, Value}; @@ -72,13 +72,11 @@ struct NestedErrorMessage { /// named `name`/`message`), and we want OTel to record the ManualFailure /// rather than the user's sibling fields. fn extract_error_message(raw: &str) -> Option { - let nested = serde_json::from_str::(raw) - .ok() - .map(|n| n.error); + let nested = parse_result_object::(raw).map(|n| n.error); if matches!(&nested, Some(em) if em.name == MANUAL_FAILURE_ERROR_NAME) { return nested; } - if let Ok(em) = serde_json::from_str::(raw) { + if let Some(em) = parse_result_object::(raw) { return Some(em); } nested diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 467b8dbf19..56b4b26fb0 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -4645,7 +4645,7 @@ pub async fn handle_queued_job( let cached_res_path = if job.cache_ttl.is_some() { match conn { Connection::Sql(db) => { - Some(cached_result_path(db, &client, &job, preview_data.as_ref()).await) + Some(cached_result_path(db, &client, &job, preview_data.as_ref()).await?) } Connection::Http(_) => None, } diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index cbda50c7bc..ed543bd517 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -2005,7 +2005,8 @@ pub async fn update_flow_status_after_job_completion_internal( if flow_job.cache_ttl.is_some() && success { let flow = RawData::Flow(flow_data.clone()); - let cached_res_path = cached_result_path(db, client, &flow_job, Some(&flow)).await; + let cached_res_path = + cached_result_path(db, client, &flow_job, Some(&flow)).await?; save_in_cache( db, diff --git a/cli/src/guidance/skills.gen.ts b/cli/src/guidance/skills.gen.ts index acedfa982b..f58092fdac 100644 --- a/cli/src/guidance/skills.gen.ts +++ b/cli/src/guidance/skills.gen.ts @@ -4608,6 +4608,14 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]] # it grows with both the width of the fan-out and \`\`attempts\`\`. Retries with # no \`\`delay\`\` all go out in a single round. # +# \`\`cache_ttl\`\` serves a previous result of the task for that many seconds +# instead of running it again. A task is keyed on its step key (its name and +# call order) and the workflow's input, not on the arguments it is called +# with, so cache one only when whether it runs, and what it receives, follow +# from the workflow's input alone. A \`\`task_script\`\` target is keyed on the +# arguments it is called with. It has no effect on a \`\`task_flow\`\` target, +# which keeps its flow's own cache policy. +# # Usage:: # # @task @@ -6012,8 +6020,8 @@ export async function main(user_id: string) { const users = await sql\`SELECT * FROM users WHERE active = \${true}\`.fetch(); // Insert/Update - await sql\`INSERT INTO users (name, email) VALUES (\${name}, \${email})\`; - await sql\`UPDATE users SET name = \${newName} WHERE id = \${user_id}\`; + await sql\`INSERT INTO users (name, email) VALUES (\${name}, \${email})\`.execute(); + await sql\`UPDATE users SET name = \${newName} WHERE id = \${user_id}\`.execute(); return user; } @@ -6032,8 +6040,8 @@ def main(user_id: str): users = db.query('SELECT * FROM users WHERE active = $1', True).fetch() # Insert/Update - db.query('INSERT INTO users (name, email) VALUES ($1, $2)', name, email) - db.query('UPDATE users SET name = $1 WHERE id = $2', new_name, user_id) + db.query('INSERT INTO users (name, email) VALUES ($1, $2)', name, email).execute() + db.query('UPDATE users SET name = $1 WHERE id = $2', new_name, user_id).execute() return user \`\`\` @@ -6042,13 +6050,14 @@ def main(user_id: str): 1. **Check existing tables** before creating new ones — reuse beats schema growth. 2. **Use parameterized queries** — never concatenate user input into SQL. -3. **Keep runnables focused** — one function per runnable; small surface area. -4. **Use descriptive keys** — \`get_user\`, not \`a\`. -5. **Always whitelist tables** — adding a runnable that queries a new table requires the table to be in \`data.tables\` first. -6. **Mark sensitive UI with \`data-wm-no-record\`** — it is what keeps that data out of a recorded demo; passwords are handled for you. -7. **Reach for \`backendAsync\` + \`waitJob\`** for long work — never a hand-written job-polling runnable. -8. **Deploy what a path runnable points at** — a path runnable aimed at a draft fails at runtime; tell the user what needs deploying. -9. **Use \`windmill-chat\` for a chat over a chat-mode flow** — never a runnable that runs the flow and polls its stream. +3. **Terminate every datatable statement** — the tagged template and \`db.query(...)\` only build a statement. It runs when you call \`fetch\` / \`fetchOne\` / \`fetchOneScalar\` / \`execute\` (\`fetch\` / \`fetch_one\` / \`fetch_one_scalar\` / \`execute\` in Python). An INSERT or UPDATE without one writes nothing and raises nothing. Awaiting the statement itself is a no-op — it is not a promise. +4. **Keep runnables focused** — one function per runnable; small surface area. +5. **Use descriptive keys** — \`get_user\`, not \`a\`. +6. **Always whitelist tables** — adding a runnable that queries a new table requires the table to be in \`data.tables\` first. +7. **Mark sensitive UI with \`data-wm-no-record\`** — it is what keeps that data out of a recorded demo; passwords are handled for you. +8. **Reach for \`backendAsync\` + \`waitJob\`** for long work — never a hand-written job-polling runnable. +9. **Deploy what a path runnable points at** — a path runnable aimed at a draft fails at runtime; tell the user what needs deploying. +10. **Use \`windmill-chat\` for a chat over a chat-mode flow** — never a runnable that runs the flow and polls its stream. `, "triggers": `--- name: triggers @@ -6741,6 +6750,13 @@ export interface TaskRetry { export interface TaskOptions { timeout?: number; tag?: string; + /** Seconds during which a previous result of this task is served instead of + * running it again. A task written inline in the workflow is keyed on its + * step key (its name and call order) and the workflow's input, not on the + * arguments it is called with, so cache one only when whether it runs, and + * what it receives, follow from the workflow's input alone. A \`taskScript\` + * target is keyed on the arguments it is called with. It has no effect on a + * \`taskFlow\` target, which keeps its flow's own cache policy. */ cache_ttl?: number; priority?: number; concurrency_limit?: number; @@ -6932,6 +6948,14 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict # it grows with both the width of the fan-out and \`\`attempts\`\`. Retries with # no \`\`delay\`\` all go out in a single round. # +# \`\`cache_ttl\`\` serves a previous result of the task for that many seconds +# instead of running it again. A task is keyed on its step key (its name and +# call order) and the workflow's input, not on the arguments it is called +# with, so cache one only when whether it runs, and what it receives, follow +# from the workflow's input alone. A \`\`task_script\`\` target is keyed on the +# arguments it is called with. It has no effect on a \`\`task_flow\`\` target, +# which keeps its flow's own cache policy. +# # Usage:: # # @task diff --git a/docs/ai-session-backups.md b/docs/ai-session-backups.md index 0b4c550456..ffec41e182 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 @@ -154,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 @@ -172,7 +220,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 @@ -187,6 +236,91 @@ 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 sessions, counted from their last activity. Each side applies +it with its own clock against its own timestamps, so no clock is compared with another +machine's, and the two do not time the same event: the server counts the last push that +completed, a browser its last local activity, which includes reading new messages and is not +pushed. A backup swept while a browser still reads its copy comes back once that browser +writes to the session 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. +- The browser sweeps its own stores when a tab resolves the logged-in user + (`sweepExpiredSessions`, from the one `onUserChange` in `sessionState.svelte.ts`), before + that tab reads a single session. A session whose last activity is older than the retention + by the browser's clock is deleted locally, record, chats, images, attached files and + artifacts. A restored session carries the backup's time as its last activity, the storage's + clock, so it counts from the later of that and the moment it was restored here + (`restoredAt`): a browser clock ahead of the storage's never deletes a session it just + brought back. Archived sessions count like any other, and persisted unsent drafts by their + pending workspace. + + The stores are shared by the user's tabs, and each keeps copies of the sessions in memory, + so every tab holds a shared Web Lock from before it reads them until it stops using them, + and the sweep deletes only while holding that lock exclusively, requested if available: + granted exactly when no tab of the user has the sessions loaded, which is why the sweep + runs where it does and nowhere else. Nothing holds a copy of what it deletes and nothing + writes the stores meanwhile, so it deletes one record at a time and without re-reading. It + also takes the tab lock the flush and the restore take, again only if available, so neither + plans nor stages a session half deleted; like the restore, it does not run where Web Locks + do not exist. With several tabs open nothing is swept, until one of them reloads alone. + + The hold is only as good as the tabs that take it, so a tab still running a build from before + it has the sessions loaded and holds nothing. A tab loaded after that one, across a deploy, + can sweep a session the older tab has in memory, and a write there afterwards brings the + record back without its chats, which the next flush pushes. It needs a tab left open across a + deploy, a session untouched for the whole retention, and the user going back to that session + in the older tab; the next sweep deletes it again. The same window is open to the + workspace-lifecycle delete in `reconcileSessionsLifecycle`, which no lock guards at all. + + What deletes is the retention the server gives as the sweep runs, asked for under both locks + (`POST /workspaces/session_workspace_retention`, its own route rather than a field on the + lifecycle status, whose answer a tab loaded before this version still reads). Never a + remembered one: a retention raised or cleared since would otherwise delete a session that is + now within it, and a persisted unsent draft has no backup to come back from. What the sweep + keeps in localStorage decides only whether to ask again — it asks when it has asked nothing + yet, when the answer it has is a day old, or when that answer marks a session expired — so + an ordinary load costs no request at all. An answer that does not arrive within five seconds + leaves the sessions for the next load rather than delete on what this browser guessed. That + route answers for a workspace the caller can be authed into, unlike the status: a status is + what to do with the caller's own sessions, a setting is the workspace's to tell, so a + disabled membership is told nothing though its sessions still reconcile. + + Each session's record goes before its pieces, so nothing plans a push for it afterwards, + and a localStorage key written before the record and removed once every piece is gone makes + a later sweep finish a deletion that failed, unless a restore brought the session back + since. The record is deleted without the tombstone a user delete leaves, which is what lets + a restore bring it back. The session's dirty mark and sync row go with it (`sessionSwept`), + unless the row still carries a removal or a restore's staging. Nothing is sent to the storage: the local + copy's age says nothing about another device's, which may have pushed the session since, + and the server applies the rule to the backup on its own. A session swept here that the + storage still lists comes back on the next restore. + ## 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/ChangeInstanceUsername.svelte b/frontend/src/lib/components/ChangeInstanceUsername.svelte index 1fcd726cea..fce0f4e8f9 100644 --- a/frontend/src/lib/components/ChangeInstanceUsername.svelte +++ b/frontend/src/lib/components/ChangeInstanceUsername.svelte @@ -3,6 +3,7 @@ import Popover from './meltComponents/Popover.svelte' import { autoPlacement } from '@floating-ui/core' import ChangeInstanceUsernameInner from './ChangeInstanceUsernameInner.svelte' + import { AlertTriangle } from 'lucide-svelte' interface Props { email: string @@ -24,9 +25,22 @@ closeButton > {#snippet trigger()} - + {#if isConflict} + + + {/if} {/snippet} {#snippet content()} - + {#snippet titleExtra()} diff --git a/frontend/src/lib/components/SuperadminSettingsInner.svelte b/frontend/src/lib/components/SuperadminSettingsInner.svelte index 97b15fe443..0fd7991597 100644 --- a/frontend/src/lib/components/SuperadminSettingsInner.svelte +++ b/frontend/src/lib/components/SuperadminSettingsInner.svelte @@ -26,11 +26,10 @@ CheckCircle2, ExternalLink, Pencil, + Settings, UserMinus, UserPlus } from 'lucide-svelte' - import Badge from './common/badge/Badge.svelte' - import Tooltip from './Tooltip.svelte' import DropdownV2 from './DropdownV2.svelte' import Popover from './meltComponents/Popover.svelte' import ConfirmationModal from './common/confirmationModal/ConfirmationModal.svelte' @@ -155,6 +154,11 @@ loadExtJwtPage(1) let tab: string = $state('users') + let usersListShown = $derived( + tab === 'users' && + !yamlMode && + (usersSubTab === 'users' || (usersSubTab === 'ext_jwt' && extJwtTokens.length === 0)) + ) $effect(() => { tab = $instanceSettingsSelectedTab @@ -320,11 +324,14 @@
-
+ +
{#if tab === 'ai' && !yamlMode} {:else if tab === 'users' && !yamlMode} -
+
{#if !automateUsernameCreation && !isCloudHosted()}

Automatic username creation

@@ -373,7 +380,7 @@ - {#if usersSubTab === 'users' || (usersSubTab === 'ext_jwt' && extJwtTokens.length === 0)} + {#if usersListShown} {filteredUsers.length} user{filteredUsers.length !== 1 ? 's' : ''} found

-
- 50} - loadMore={50} - on:loadMore={() => { - nbDisplayed += 50 - }} - > + +
+ Email @@ -434,7 +437,7 @@ Kind {/if} Role - + Actions @@ -443,12 +446,22 @@ {#if filteredUsers && users} {#each filteredUsers.slice(0, nbDisplayed) as { email, super_admin, devops, login_type, name, username, operator_only, is_workspace_admin, role_source, disabled, workspace_id }, i (email + '::' + (workspace_id ?? ''))} {@const isServiceAccount = login_type === 'service_account'} + {@const groupRole = + role_source === 'instance_group' && (super_admin || devops)} + + {@const groupRoleTooltip = + 'Role is set by an instance group. Superadmin and Devops can be set here, but demoting to User requires removing the user from the group.'} + {@const serviceAccountTooltip = + 'Service accounts are always users in the instance. Their workspace role is managed in the workspace user settings.'} + - +
{#if isServiceAccount} @@ -458,14 +471,6 @@ >{email} {/if} - {#if workspace_id} - - {truncate(workspace_id, 20)} - - {/if} {#if disabled} {#if automateUsernameCreation} - + {#if username} {username} {:else} @@ -503,133 +508,157 @@ > {#if activeOnly} - {#if is_workspace_admin} - Admin - {:else if operator_only} - Operator only - {:else} - Developer - {/if} + + {#if is_workspace_admin} + Admin + {:else if operator_only} + Operator only + {:else} + Developer + {/if} + {/if} - {#if isServiceAccount} -
+ +
+ {#key `${super_admin}_${devops}_${role_source}`} + { + if (email == $userStore?.email) { + sendUserToast('You cannot demote yourself', true) + listUsers(activeOnly) + return + } + + let role = e.detail + + if (role === 'super_admin') { + await UserService.globalUserUpdate({ + email, + requestBody: { + is_super_admin: true, + is_devops: false + } + }) + } + if (role === 'devops') { + await UserService.globalUserUpdate({ + email, + requestBody: { + is_super_admin: false, + is_devops: true + } + }) + } + if (role === 'user') { + await UserService.globalUserUpdate({ + email, + requestBody: { + is_super_admin: false, + is_devops: false + } + }) + } + sendUserToast('User updated') + listUsers(activeOnly) + }} + > + {#snippet children({ item })} + + + + {/snippet} + + {/key} + {#if isServiceAccount} {is_workspace_admin ? 'Admin' : operator_only ? 'Operator' : 'Developer'} + in + {#if workspace_id} + closeDrawer?.()} + >{truncate(workspace_id, 20)} + {:else} + its workspace + {/if} - - Service-account role is managed in the workspace user settings. - -
- {:else} -
- {#key `${super_admin}_${devops}_${role_source}`} - { - if (email == $userStore?.email) { - sendUserToast('You cannot demote yourself', true) - listUsers(activeOnly) - return - } - - let role = e.detail - - if (role === 'super_admin') { - await UserService.globalUserUpdate({ - email, - requestBody: { - is_super_admin: true, - is_devops: false - } - }) - } - if (role === 'devops') { - await UserService.globalUserUpdate({ - email, - requestBody: { - is_super_admin: false, - is_devops: true - } - }) - } - if (role === 'user') { - await UserService.globalUserUpdate({ - email, - requestBody: { - is_super_admin: false, - is_devops: false - } - }) - } - sendUserToast('User updated') - listUsers(activeOnly) - }} - > - {#snippet children({ item })} - - - - {/snippet} - - {/key} - {#if role_source === 'instance_group' && (super_admin || devops)} - closeDrawer?.()} - > - Set by instance group - - {/if} -
- {/if} + {:else if groupRole} + closeDrawer?.()} + > + Set by instance group + + {/if} +
- +
{#if isServiceAccount} {#if workspace_id} - Manage in workspace + closeDrawer?.(), + href: `${base}/workspace_settings?tab=users&workspace=${workspace_id}` + } + ]} + /> {/if} {:else}
{/each} + {#if filteredUsers.length > nbDisplayed} + {@const remaining = Math.min(50, filteredUsers.length - nbDisplayed)} + + + + + + + {/if} {/if} diff --git a/frontend/src/lib/components/common/toggleButton-v2/ToggleButton.svelte b/frontend/src/lib/components/common/toggleButton-v2/ToggleButton.svelte index 621e8383d0..2773503df9 100644 --- a/frontend/src/lib/components/common/toggleButton-v2/ToggleButton.svelte +++ b/frontend/src/lib/components/common/toggleButton-v2/ToggleButton.svelte @@ -7,6 +7,9 @@ interface Props { label?: string | undefined + /** Shown instead of `label` below the `xl` breakpoint, for groups that must keep + * their width inside a narrow table cell. The full label stays the accessible name. */ + shortLabel?: string | undefined iconOnly?: boolean tooltip?: string | undefined icon?: any | undefined @@ -30,6 +33,7 @@ let { label = undefined, + shortLabel = undefined, iconOnly = false, tooltip = undefined, icon = undefined, @@ -68,6 +72,7 @@
- +
diff --git a/python-client/wmill/wmill/client.py b/python-client/wmill/wmill/client.py index 918a16f2b0..dbc9d9b193 100644 --- a/python-client/wmill/wmill/client.py +++ b/python-client/wmill/wmill/client.py @@ -3327,6 +3327,14 @@ def task( it grows with both the width of the fan-out and ``attempts``. Retries with no ``delay`` all go out in a single round. + ``cache_ttl`` serves a previous result of the task for that many seconds + instead of running it again. A task is keyed on its step key (its name and + call order) and the workflow's input, not on the arguments it is called + with, so cache one only when whether it runs, and what it receives, follow + from the workflow's input alone. A ``task_script`` target is keyed on the + arguments it is called with. It has no effect on a ``task_flow`` target, + which keeps its flow's own cache policy. + Usage:: @task diff --git a/system_prompts/auto-generated/prompts.ts b/system_prompts/auto-generated/prompts.ts index 4b4552dca4..1e92c8e69b 100644 --- a/system_prompts/auto-generated/prompts.ts +++ b/system_prompts/auto-generated/prompts.ts @@ -920,8 +920,8 @@ export async function main(user_id: string) { const users = await sql\`SELECT * FROM users WHERE active = \${true}\`.fetch(); // Insert/Update - await sql\`INSERT INTO users (name, email) VALUES (\${name}, \${email})\`; - await sql\`UPDATE users SET name = \${newName} WHERE id = \${user_id}\`; + await sql\`INSERT INTO users (name, email) VALUES (\${name}, \${email})\`.execute(); + await sql\`UPDATE users SET name = \${newName} WHERE id = \${user_id}\`.execute(); return user; } @@ -940,8 +940,8 @@ def main(user_id: str): users = db.query('SELECT * FROM users WHERE active = $1', True).fetch() # Insert/Update - db.query('INSERT INTO users (name, email) VALUES ($1, $2)', name, email) - db.query('UPDATE users SET name = $1 WHERE id = $2', new_name, user_id) + db.query('INSERT INTO users (name, email) VALUES ($1, $2)', name, email).execute() + db.query('UPDATE users SET name = $1 WHERE id = $2', new_name, user_id).execute() return user \`\`\` @@ -950,13 +950,14 @@ def main(user_id: str): 1. **Check existing tables** before creating new ones — reuse beats schema growth. 2. **Use parameterized queries** — never concatenate user input into SQL. -3. **Keep runnables focused** — one function per runnable; small surface area. -4. **Use descriptive keys** — \`get_user\`, not \`a\`. -5. **Always whitelist tables** — adding a runnable that queries a new table requires the table to be in \`data.tables\` first. -6. **Mark sensitive UI with \`data-wm-no-record\`** — it is what keeps that data out of a recorded demo; passwords are handled for you. -7. **Reach for \`backendAsync\` + \`waitJob\`** for long work — never a hand-written job-polling runnable. -8. **Deploy what a path runnable points at** — a path runnable aimed at a draft fails at runtime; tell the user what needs deploying. -9. **Use \`windmill-chat\` for a chat over a chat-mode flow** — never a runnable that runs the flow and polls its stream. +3. **Terminate every datatable statement** — the tagged template and \`db.query(...)\` only build a statement. It runs when you call \`fetch\` / \`fetchOne\` / \`fetchOneScalar\` / \`execute\` (\`fetch\` / \`fetch_one\` / \`fetch_one_scalar\` / \`execute\` in Python). An INSERT or UPDATE without one writes nothing and raises nothing. Awaiting the statement itself is a no-op — it is not a promise. +4. **Keep runnables focused** — one function per runnable; small surface area. +5. **Use descriptive keys** — \`get_user\`, not \`a\`. +6. **Always whitelist tables** — adding a runnable that queries a new table requires the table to be in \`data.tables\` first. +7. **Mark sensitive UI with \`data-wm-no-record\`** — it is what keeps that data out of a recorded demo; passwords are handled for you. +8. **Reach for \`backendAsync\` + \`waitJob\`** for long work — never a hand-written job-polling runnable. +9. **Deploy what a path runnable points at** — a path runnable aimed at a draft fails at runtime; tell the user what needs deploying. +10. **Use \`windmill-chat\` for a chat over a chat-mode flow** — never a runnable that runs the flow and polls its stream. `; export const PIPELINE_BASE = `# Data pipeline authoring @@ -2587,6 +2588,14 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]] # it grows with both the width of the fan-out and \`\`attempts\`\`. Retries with # no \`\`delay\`\` all go out in a single round. # +# \`\`cache_ttl\`\` serves a previous result of the task for that many seconds +# instead of running it again. A task is keyed on its step key (its name and +# call order) and the workflow's input, not on the arguments it is called +# with, so cache one only when whether it runs, and what it receives, follow +# from the workflow's input alone. A \`\`task_script\`\` target is keyed on the +# arguments it is called with. It has no effect on a \`\`task_flow\`\` target, +# which keeps its flow's own cache policy. +# # Usage:: # # @task @@ -2737,6 +2746,13 @@ export interface TaskRetry { export interface TaskOptions { timeout?: number; tag?: string; + /** Seconds during which a previous result of this task is served instead of + * running it again. A task written inline in the workflow is keyed on its + * step key (its name and call order) and the workflow's input, not on the + * arguments it is called with, so cache one only when whether it runs, and + * what it receives, follow from the workflow's input alone. A \`taskScript\` + * target is keyed on the arguments it is called with. It has no effect on a + * \`taskFlow\` target, which keeps its flow's own cache policy. */ cache_ttl?: number; priority?: number; concurrency_limit?: number; @@ -2928,6 +2944,14 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict # it grows with both the width of the fan-out and \`\`attempts\`\`. Retries with # no \`\`delay\`\` all go out in a single round. # +# \`\`cache_ttl\`\` serves a previous result of the task for that many seconds +# instead of running it again. A task is keyed on its step key (its name and +# call order) and the workflow's input, not on the arguments it is called +# with, so cache one only when whether it runs, and what it receives, follow +# from the workflow's input alone. A \`\`task_script\`\` target is keyed on the +# arguments it is called with. It has no effect on a \`\`task_flow\`\` target, +# which keeps its flow's own cache policy. +# # Usage:: # # @task diff --git a/system_prompts/auto-generated/script.md b/system_prompts/auto-generated/script.md index 7883a8ba4b..f278a67fba 100644 --- a/system_prompts/auto-generated/script.md +++ b/system_prompts/auto-generated/script.md @@ -2728,6 +2728,14 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]] # it grows with both the width of the fan-out and ``attempts``. Retries with # no ``delay`` all go out in a single round. # +# ``cache_ttl`` serves a previous result of the task for that many seconds +# instead of running it again. A task is keyed on its step key (its name and +# call order) and the workflow's input, not on the arguments it is called +# with, so cache one only when whether it runs, and what it receives, follow +# from the workflow's input alone. A ``task_script`` target is keyed on the +# arguments it is called with. It has no effect on a ``task_flow`` target, +# which keeps its flow's own cache policy. +# # Usage:: # # @task diff --git a/system_prompts/auto-generated/sdks/python.md b/system_prompts/auto-generated/sdks/python.md index 4aacf4d1b4..4f727c7d45 100644 --- a/system_prompts/auto-generated/sdks/python.md +++ b/system_prompts/auto-generated/sdks/python.md @@ -672,6 +672,14 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]] # it grows with both the width of the fan-out and ``attempts``. Retries with # no ``delay`` all go out in a single round. # +# ``cache_ttl`` serves a previous result of the task for that many seconds +# instead of running it again. A task is keyed on its step key (its name and +# call order) and the workflow's input, not on the arguments it is called +# with, so cache one only when whether it runs, and what it receives, follow +# from the workflow's input alone. A ``task_script`` target is keyed on the +# arguments it is called with. It has no effect on a ``task_flow`` target, +# which keeps its flow's own cache policy. +# # Usage:: # # @task diff --git a/system_prompts/auto-generated/sdks/wac-python.md b/system_prompts/auto-generated/sdks/wac-python.md index 816ea4959b..a8e98a9d4f 100644 --- a/system_prompts/auto-generated/sdks/wac-python.md +++ b/system_prompts/auto-generated/sdks/wac-python.md @@ -58,6 +58,14 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict # it grows with both the width of the fan-out and ``attempts``. Retries with # no ``delay`` all go out in a single round. # +# ``cache_ttl`` serves a previous result of the task for that many seconds +# instead of running it again. A task is keyed on its step key (its name and +# call order) and the workflow's input, not on the arguments it is called +# with, so cache one only when whether it runs, and what it receives, follow +# from the workflow's input alone. A ``task_script`` target is keyed on the +# arguments it is called with. It has no effect on a ``task_flow`` target, +# which keeps its flow's own cache policy. +# # Usage:: # # @task diff --git a/system_prompts/auto-generated/sdks/wac-typescript.md b/system_prompts/auto-generated/sdks/wac-typescript.md index 66eecfe761..608d75d18c 100644 --- a/system_prompts/auto-generated/sdks/wac-typescript.md +++ b/system_prompts/auto-generated/sdks/wac-typescript.md @@ -34,6 +34,13 @@ export interface TaskRetry { export interface TaskOptions { timeout?: number; tag?: string; + /** Seconds during which a previous result of this task is served instead of + * running it again. A task written inline in the workflow is keyed on its + * step key (its name and call order) and the workflow's input, not on the + * arguments it is called with, so cache one only when whether it runs, and + * what it receives, follow from the workflow's input alone. A `taskScript` + * target is keyed on the arguments it is called with. It has no effect on a + * `taskFlow` target, which keeps its flow's own cache policy. */ cache_ttl?: number; priority?: number; concurrency_limit?: number; diff --git a/system_prompts/auto-generated/skills/raw-app/SKILL.md b/system_prompts/auto-generated/skills/raw-app/SKILL.md index da4a9729be..ed79172b9d 100644 --- a/system_prompts/auto-generated/skills/raw-app/SKILL.md +++ b/system_prompts/auto-generated/skills/raw-app/SKILL.md @@ -435,8 +435,8 @@ export async function main(user_id: string) { const users = await sql`SELECT * FROM users WHERE active = ${true}`.fetch(); // Insert/Update - await sql`INSERT INTO users (name, email) VALUES (${name}, ${email})`; - await sql`UPDATE users SET name = ${newName} WHERE id = ${user_id}`; + await sql`INSERT INTO users (name, email) VALUES (${name}, ${email})`.execute(); + await sql`UPDATE users SET name = ${newName} WHERE id = ${user_id}`.execute(); return user; } @@ -455,8 +455,8 @@ def main(user_id: str): users = db.query('SELECT * FROM users WHERE active = $1', True).fetch() # Insert/Update - db.query('INSERT INTO users (name, email) VALUES ($1, $2)', name, email) - db.query('UPDATE users SET name = $1 WHERE id = $2', new_name, user_id) + db.query('INSERT INTO users (name, email) VALUES ($1, $2)', name, email).execute() + db.query('UPDATE users SET name = $1 WHERE id = $2', new_name, user_id).execute() return user ``` @@ -465,10 +465,11 @@ def main(user_id: str): 1. **Check existing tables** before creating new ones — reuse beats schema growth. 2. **Use parameterized queries** — never concatenate user input into SQL. -3. **Keep runnables focused** — one function per runnable; small surface area. -4. **Use descriptive keys** — `get_user`, not `a`. -5. **Always whitelist tables** — adding a runnable that queries a new table requires the table to be in `data.tables` first. -6. **Mark sensitive UI with `data-wm-no-record`** — it is what keeps that data out of a recorded demo; passwords are handled for you. -7. **Reach for `backendAsync` + `waitJob`** for long work — never a hand-written job-polling runnable. -8. **Deploy what a path runnable points at** — a path runnable aimed at a draft fails at runtime; tell the user what needs deploying. -9. **Use `windmill-chat` for a chat over a chat-mode flow** — never a runnable that runs the flow and polls its stream. +3. **Terminate every datatable statement** — the tagged template and `db.query(...)` only build a statement. It runs when you call `fetch` / `fetchOne` / `fetchOneScalar` / `execute` (`fetch` / `fetch_one` / `fetch_one_scalar` / `execute` in Python). An INSERT or UPDATE without one writes nothing and raises nothing. Awaiting the statement itself is a no-op — it is not a promise. +4. **Keep runnables focused** — one function per runnable; small surface area. +5. **Use descriptive keys** — `get_user`, not `a`. +6. **Always whitelist tables** — adding a runnable that queries a new table requires the table to be in `data.tables` first. +7. **Mark sensitive UI with `data-wm-no-record`** — it is what keeps that data out of a recorded demo; passwords are handled for you. +8. **Reach for `backendAsync` + `waitJob`** for long work — never a hand-written job-polling runnable. +9. **Deploy what a path runnable points at** — a path runnable aimed at a draft fails at runtime; tell the user what needs deploying. +10. **Use `windmill-chat` for a chat over a chat-mode flow** — never a runnable that runs the flow and polls its stream. diff --git a/system_prompts/auto-generated/skills/write-script-python3/SKILL.md b/system_prompts/auto-generated/skills/write-script-python3/SKILL.md index 8100eebc25..1ccb376a9f 100644 --- a/system_prompts/auto-generated/skills/write-script-python3/SKILL.md +++ b/system_prompts/auto-generated/skills/write-script-python3/SKILL.md @@ -857,6 +857,14 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]] # it grows with both the width of the fan-out and ``attempts``. Retries with # no ``delay`` all go out in a single round. # +# ``cache_ttl`` serves a previous result of the task for that many seconds +# instead of running it again. A task is keyed on its step key (its name and +# call order) and the workflow's input, not on the arguments it is called +# with, so cache one only when whether it runs, and what it receives, follow +# from the workflow's input alone. A ``task_script`` target is keyed on the +# arguments it is called with. It has no effect on a ``task_flow`` target, +# which keeps its flow's own cache policy. +# # Usage:: # # @task diff --git a/system_prompts/auto-generated/skills/write-workflow-as-code/SKILL.md b/system_prompts/auto-generated/skills/write-workflow-as-code/SKILL.md index c41bd54f04..8e5ab1f06c 100644 --- a/system_prompts/auto-generated/skills/write-workflow-as-code/SKILL.md +++ b/system_prompts/auto-generated/skills/write-workflow-as-code/SKILL.md @@ -277,6 +277,13 @@ export interface TaskRetry { export interface TaskOptions { timeout?: number; tag?: string; + /** Seconds during which a previous result of this task is served instead of + * running it again. A task written inline in the workflow is keyed on its + * step key (its name and call order) and the workflow's input, not on the + * arguments it is called with, so cache one only when whether it runs, and + * what it receives, follow from the workflow's input alone. A `taskScript` + * target is keyed on the arguments it is called with. It has no effect on a + * `taskFlow` target, which keeps its flow's own cache policy. */ cache_ttl?: number; priority?: number; concurrency_limit?: number; @@ -468,6 +475,14 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict # it grows with both the width of the fan-out and ``attempts``. Retries with # no ``delay`` all go out in a single round. # +# ``cache_ttl`` serves a previous result of the task for that many seconds +# instead of running it again. A task is keyed on its step key (its name and +# call order) and the workflow's input, not on the arguments it is called +# with, so cache one only when whether it runs, and what it receives, follow +# from the workflow's input alone. A ``task_script`` target is keyed on the +# arguments it is called with. It has no effect on a ``task_flow`` target, +# which keeps its flow's own cache policy. +# # Usage:: # # @task diff --git a/system_prompts/base/raw-app.md b/system_prompts/base/raw-app.md index 3849a465a0..f7eaef944d 100644 --- a/system_prompts/base/raw-app.md +++ b/system_prompts/base/raw-app.md @@ -200,8 +200,8 @@ export async function main(user_id: string) { const users = await sql`SELECT * FROM users WHERE active = ${true}`.fetch(); // Insert/Update - await sql`INSERT INTO users (name, email) VALUES (${name}, ${email})`; - await sql`UPDATE users SET name = ${newName} WHERE id = ${user_id}`; + await sql`INSERT INTO users (name, email) VALUES (${name}, ${email})`.execute(); + await sql`UPDATE users SET name = ${newName} WHERE id = ${user_id}`.execute(); return user; } @@ -220,8 +220,8 @@ def main(user_id: str): users = db.query('SELECT * FROM users WHERE active = $1', True).fetch() # Insert/Update - db.query('INSERT INTO users (name, email) VALUES ($1, $2)', name, email) - db.query('UPDATE users SET name = $1 WHERE id = $2', new_name, user_id) + db.query('INSERT INTO users (name, email) VALUES ($1, $2)', name, email).execute() + db.query('UPDATE users SET name = $1 WHERE id = $2', new_name, user_id).execute() return user ``` @@ -230,10 +230,11 @@ def main(user_id: str): 1. **Check existing tables** before creating new ones — reuse beats schema growth. 2. **Use parameterized queries** — never concatenate user input into SQL. -3. **Keep runnables focused** — one function per runnable; small surface area. -4. **Use descriptive keys** — `get_user`, not `a`. -5. **Always whitelist tables** — adding a runnable that queries a new table requires the table to be in `data.tables` first. -6. **Mark sensitive UI with `data-wm-no-record`** — it is what keeps that data out of a recorded demo; passwords are handled for you. -7. **Reach for `backendAsync` + `waitJob`** for long work — never a hand-written job-polling runnable. -8. **Deploy what a path runnable points at** — a path runnable aimed at a draft fails at runtime; tell the user what needs deploying. -9. **Use `windmill-chat` for a chat over a chat-mode flow** — never a runnable that runs the flow and polls its stream. +3. **Terminate every datatable statement** — the tagged template and `db.query(...)` only build a statement. It runs when you call `fetch` / `fetchOne` / `fetchOneScalar` / `execute` (`fetch` / `fetch_one` / `fetch_one_scalar` / `execute` in Python). An INSERT or UPDATE without one writes nothing and raises nothing. Awaiting the statement itself is a no-op — it is not a promise. +4. **Keep runnables focused** — one function per runnable; small surface area. +5. **Use descriptive keys** — `get_user`, not `a`. +6. **Always whitelist tables** — adding a runnable that queries a new table requires the table to be in `data.tables` first. +7. **Mark sensitive UI with `data-wm-no-record`** — it is what keeps that data out of a recorded demo; passwords are handled for you. +8. **Reach for `backendAsync` + `waitJob`** for long work — never a hand-written job-polling runnable. +9. **Deploy what a path runnable points at** — a path runnable aimed at a draft fails at runtime; tell the user what needs deploying. +10. **Use `windmill-chat` for a chat over a chat-mode flow** — never a runnable that runs the flow and polls its stream. diff --git a/typescript-client/client.ts b/typescript-client/client.ts index 7b10c732be..dd2123e746 100644 --- a/typescript-client/client.ts +++ b/typescript-client/client.ts @@ -1712,6 +1712,13 @@ export interface TaskRetry { export interface TaskOptions { timeout?: number; tag?: string; + /** Seconds during which a previous result of this task is served instead of + * running it again. A task written inline in the workflow is keyed on its + * step key (its name and call order) and the workflow's input, not on the + * arguments it is called with, so cache one only when whether it runs, and + * what it receives, follow from the workflow's input alone. A `taskScript` + * target is keyed on the arguments it is called with. It has no effect on a + * `taskFlow` target, which keeps its flow's own cache policy. */ cache_ttl?: number; priority?: number; concurrency_limit?: number;