mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-25 08:03:45 +00:00
perf: evict the address cache on change so app dispatch can read it
This commit is contained in:
@@ -0,0 +1,6 @@
|
||||
-- Add down migration script here
|
||||
DROP TRIGGER IF EXISTS password_superadmin_email_change_trigger ON password;
|
||||
DROP TRIGGER IF EXISTS usr_email_update_trigger ON usr;
|
||||
DROP TRIGGER IF EXISTS usr_email_change_trigger ON usr;
|
||||
DROP FUNCTION IF EXISTS notify_superadmin_email_change();
|
||||
DROP FUNCTION IF EXISTS notify_usr_email_change();
|
||||
@@ -0,0 +1,55 @@
|
||||
-- Add up migration script here
|
||||
-- Emit a notify_event so every server evicts its cached `permissioned_as` -> address mapping
|
||||
-- (windmill-common EMAIL_CACHE). That address is derived at job dispatch and feeds the
|
||||
-- instance-superadmin check and `email_to_igroup`, so a replica serving a stale one runs jobs
|
||||
-- with the wrong authorization until the TTL expires. SECURITY DEFINER so the INSERT runs as the
|
||||
-- function owner rather than the invoking windmill_user/windmill_admin role, matching the other
|
||||
-- notify_* triggers.
|
||||
CREATE OR REPLACE FUNCTION notify_usr_email_change()
|
||||
RETURNS TRIGGER AS $$
|
||||
BEGIN
|
||||
INSERT INTO notify_event (channel, payload)
|
||||
VALUES (
|
||||
'notify_user_email_change',
|
||||
COALESCE(NEW.workspace_id, OLD.workspace_id) || ':' || COALESCE(NEW.username, OLD.username)
|
||||
);
|
||||
-- A rename leaves the OLD username cached against this account's address; evict both keys.
|
||||
IF TG_OP = 'UPDATE' AND NEW.username IS DISTINCT FROM OLD.username THEN
|
||||
INSERT INTO notify_event (channel, payload)
|
||||
VALUES ('notify_user_email_change', OLD.workspace_id || ':' || OLD.username);
|
||||
END IF;
|
||||
RETURN COALESCE(NEW, OLD);
|
||||
END;
|
||||
$$ LANGUAGE plpgsql SECURITY DEFINER;
|
||||
|
||||
-- INSERT matters too: a lookup that resolved to nobody is cached as the synthetic
|
||||
-- `{username}@unknown.windmill.dev`, so creating the row has to drop that entry.
|
||||
CREATE TRIGGER usr_email_change_trigger
|
||||
AFTER INSERT OR DELETE ON usr
|
||||
FOR EACH ROW
|
||||
EXECUTE FUNCTION notify_usr_email_change();
|
||||
|
||||
CREATE TRIGGER usr_email_update_trigger
|
||||
AFTER UPDATE OF email, username ON usr
|
||||
FOR EACH ROW
|
||||
WHEN (OLD.email IS DISTINCT FROM NEW.email OR OLD.username IS DISTINCT FROM NEW.username)
|
||||
EXECUTE FUNCTION notify_usr_email_change();
|
||||
|
||||
-- A superadmin acting outside their workspaces resolves through `password` instead, and that
|
||||
-- row names no workspace, so there is no key to target: clear the whole cache. The empty payload
|
||||
-- is the wildcard. Confined to superadmins because they are the only accounts the `usr` triggers
|
||||
-- above cannot cover.
|
||||
CREATE OR REPLACE FUNCTION notify_superadmin_email_change()
|
||||
RETURNS TRIGGER AS $$
|
||||
BEGIN
|
||||
INSERT INTO notify_event (channel, payload) VALUES ('notify_user_email_change', '');
|
||||
RETURN COALESCE(NEW, OLD);
|
||||
END;
|
||||
$$ LANGUAGE plpgsql SECURITY DEFINER;
|
||||
|
||||
CREATE TRIGGER password_superadmin_email_change_trigger
|
||||
AFTER UPDATE OF email, username ON password
|
||||
FOR EACH ROW
|
||||
WHEN (NEW.super_admin
|
||||
AND (OLD.email IS DISTINCT FROM NEW.email OR OLD.username IS DISTINCT FROM NEW.username))
|
||||
EXECUTE FUNCTION notify_superadmin_email_change();
|
||||
@@ -1831,6 +1831,17 @@ async fn process_notify_event(
|
||||
);
|
||||
windmill_api::auth::invalidate_token_from_cache(payload);
|
||||
}
|
||||
"notify_user_email_change" => {
|
||||
// `<workspace_id>:<username>`; workspace ids can't contain ':'. Empty payload is the
|
||||
// wildcard a `password` change emits, having no workspace to name.
|
||||
if payload.is_empty() {
|
||||
tracing::info!("Superadmin email change detected, clearing address cache");
|
||||
windmill_common::users::clear_email_cache();
|
||||
} else if let Some((workspace_id, username)) = payload.split_once(':') {
|
||||
tracing::info!("User email change detected, invalidating cache: {payload}");
|
||||
windmill_common::users::invalidate_email_cache(workspace_id, username);
|
||||
}
|
||||
}
|
||||
"notify_app_policy_change" => {
|
||||
// payload is `<workspace_id>:<path>`; workspace ids can't contain ':'.
|
||||
if server_mode {
|
||||
|
||||
@@ -4860,13 +4860,12 @@ fn audited_on_behalf_of(policy: &Policy, authed: &ApiAuthed) -> Option<String> {
|
||||
|
||||
/// The identity an anonymous or publisher execution runs as.
|
||||
///
|
||||
/// The address is resolved **uncached**, and that is not optional: it becomes the job's
|
||||
/// `permissioned_as_email`, from which `fetch_authed_from_permissioned_as` derives the
|
||||
/// superadmin flag and the instance groups. `EMAIL_CACHE` has a 60s TTL and nothing evicts it,
|
||||
/// so a cached read on a replica that did not handle an email change would enqueue jobs
|
||||
/// carrying the pre-change authorization. It costs one indexed lookup per execution, against a
|
||||
/// request that already does far more — and only for a `u/` principal, since `g/` and bare
|
||||
/// addresses resolve without touching the database.
|
||||
/// The address becomes the job's `permissioned_as_email`, from which
|
||||
/// `fetch_authed_from_permissioned_as` derives the superadmin flag and the instance groups, so it
|
||||
/// must not be stale. It is read through `EMAIL_CACHE` anyway: the `notify_user_email_change`
|
||||
/// trigger evicts the key on every change that can move it, on every replica, the same way this
|
||||
/// route's own `APP_POLICY_CACHE` is kept honest. Resolving per execution instead would put the
|
||||
/// only uncached query on a path that otherwise serves entirely from cache.
|
||||
async fn get_on_behalf_of(policy: &Policy, w_id: &str, db: &DB) -> Result<(String, String)> {
|
||||
let permissioned_as = policy
|
||||
.on_behalf_of
|
||||
@@ -4878,12 +4877,8 @@ async fn get_on_behalf_of(policy: &Policy, w_id: &str, db: &DB) -> Result<(Strin
|
||||
)
|
||||
})?
|
||||
.to_string();
|
||||
let email = windmill_common::users::get_email_from_permissioned_as_uncached(
|
||||
&permissioned_as,
|
||||
w_id,
|
||||
db,
|
||||
)
|
||||
.await?;
|
||||
let email =
|
||||
windmill_common::users::get_email_from_permissioned_as(&permissioned_as, w_id, db).await?;
|
||||
Ok((permissioned_as, email))
|
||||
}
|
||||
|
||||
|
||||
@@ -159,6 +159,12 @@ pub fn invalidate_email_cache(workspace_id: &str, username: &str) {
|
||||
EMAIL_CACHE.remove(&(workspace_id.to_string(), username.to_string()));
|
||||
}
|
||||
|
||||
/// Drop every entry, for the changes that cannot name the key they invalidate: a superadmin
|
||||
/// resolves through `password`, whose row names no workspace.
|
||||
pub fn clear_email_cache() {
|
||||
EMAIL_CACHE.clear();
|
||||
}
|
||||
|
||||
/// Inverse of [`get_email_from_permissioned_as`]: the principal an on-behalf-of email
|
||||
/// names in this workspace, for callers that supply the email alone.
|
||||
///
|
||||
|
||||
@@ -360,6 +360,45 @@ async fn test_trigger_notify_workspace_key_change(db: Pool<Postgres>) {
|
||||
);
|
||||
}
|
||||
|
||||
/// The address a job runs as is served from a process-local cache, so every change that can move
|
||||
/// a `(workspace, username)` -> email mapping has to reach the other replicas as an eviction.
|
||||
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
|
||||
async fn test_trigger_notify_user_email_change(db: Pool<Postgres>) {
|
||||
let before_id = get_latest_event_id(&db).await.unwrap();
|
||||
|
||||
sqlx::query("UPDATE usr SET email = 'renamed@windmill.dev' WHERE workspace_id = 'test-workspace' AND username = 'test-user'")
|
||||
.execute(&db)
|
||||
.await
|
||||
.expect("Failed to change email");
|
||||
|
||||
let events = poll_notify_events(&db, before_id)
|
||||
.await
|
||||
.expect("Should poll events");
|
||||
assert!(
|
||||
events.iter().any(|e| e.channel == "notify_user_email_change"
|
||||
&& e.payload == "test-workspace:test-user"),
|
||||
"email change should evict the key it moved"
|
||||
);
|
||||
|
||||
// A superadmin outside their workspaces resolves through `password`, which names no
|
||||
// workspace: the wildcard is the only way to reach that key.
|
||||
let before_id = get_latest_event_id(&db).await.unwrap();
|
||||
sqlx::query("UPDATE password SET email = 'sa2@windmill.dev' WHERE email = 'test@windmill.dev'")
|
||||
.execute(&db)
|
||||
.await
|
||||
.expect("Failed to change superadmin email");
|
||||
|
||||
let events = poll_notify_events(&db, before_id)
|
||||
.await
|
||||
.expect("Should poll events");
|
||||
assert!(
|
||||
events
|
||||
.iter()
|
||||
.any(|e| e.channel == "notify_user_email_change" && e.payload.is_empty()),
|
||||
"superadmin email change should emit the wildcard"
|
||||
);
|
||||
}
|
||||
|
||||
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
|
||||
async fn test_trigger_notify_token_invalidation(db: Pool<Postgres>) {
|
||||
// First insert a session token with token_hash and token_prefix
|
||||
|
||||
Reference in New Issue
Block a user