fix: keep persisted addresses off the cache the poller evicts asynchronously

backend/windmill-api-flows/src/flows.rs

@@ -891,7 +891,7 @@ async fn derived_on_behalf_of_email(
     };
     // Uncached, for the reason given on `prefetch_cached_script`: this pair is round-tripped.
     Ok(Some(
-        windmill_common::users::get_email_from_permissioned_as(permissioned_as, w_id, db)
+        windmill_common::users::get_email_from_permissioned_as_uncached(permissioned_as, w_id, db)
             .await?,
     ))
 }

backend/windmill-api/src/apps.rs

@@ -4671,7 +4671,7 @@ async fn stored_on_behalf_of_email(
         return Ok(None);
     };
     Ok(Some(
-        windmill_common::users::get_email_from_permissioned_as(permissioned_as, w_id, db)
+        windmill_common::users::get_email_from_permissioned_as_uncached(permissioned_as, w_id, db)
             .await?,
     ))
 }

backend/windmill-api/src/workspaces_export.rs

@@ -149,7 +149,7 @@ async fn derive_email(
     }
     // The memo above holds this to one query per distinct principal per export.
     let email =
-        windmill_common::users::get_email_from_permissioned_as(permissioned_as, w_id, db)
+        windmill_common::users::get_email_from_permissioned_as_uncached(permissioned_as, w_id, db)
             .await?;
     cache.insert(permissioned_as.to_string(), email.clone());
     Ok(Some(email))

backend/windmill-common/src/folders.rs

@@ -79,7 +79,7 @@ pub async fn resolve_folder_default_on_behalf_of(
         return Ok(None);
     };
     let email =
-        crate::users::get_email_from_permissioned_as(&permissioned_as, w_id, db).await?;
+        crate::users::get_email_from_permissioned_as_uncached(&permissioned_as, w_id, db).await?;
     Ok(Some((email, permissioned_as)))
 }
 

backend/windmill-common/src/lib.rs

@@ -234,7 +234,7 @@ pub async fn resolve_on_behalf_of(
             // workspace's principal beside another's address.
             if let Some(email) = on_behalf_of_email {
                 let named =
-                    users::get_email_from_permissioned_as(permissioned_as, w_id, db)
+                    users::get_email_from_permissioned_as_uncached(permissioned_as, w_id, db)
                         .await?;
                 if named != email {
                     return Err(Error::BadRequest(format!(
@@ -1684,7 +1684,7 @@ pub async fn legacy_on_behalf_of_email(
         return Ok(None);
     }
     Ok(Some(
-        users::get_email_from_permissioned_as(permissioned_as, w_id, db).await?,
+        users::get_email_from_permissioned_as_uncached(permissioned_as, w_id, db).await?,
     ))
 }
 

backend/windmill-common/src/scripts.rs

@@ -92,7 +92,7 @@ async fn prefetch_cached_script_inner(
 ) -> crate::error::Result<Script<ScriptRunnableSettingsInline>> {
     let derived_email = match script.on_behalf_of.as_deref().filter(|_| derive_email) {
         Some(permissioned_as) => Some(
-            crate::users::get_email_from_permissioned_as(
+            crate::users::get_email_from_permissioned_as_uncached(
                 permissioned_as,
                 &script.workspace_id,
                 db,

backend/windmill-common/src/users.rs

@@ -234,12 +234,39 @@ pub async fn get_email_from_permissioned_as<'c>(
     permissioned_as: &str,
     workspace_id: &str,
     db: impl sqlx::PgExecutor<'c>,
+) -> crate::error::Result<String> {
+    get_email_from_permissioned_as_inner(permissioned_as, workspace_id, db, true).await
+}
+
+/// [`get_email_from_permissioned_as`] for a value about to be **persisted**.
+///
+/// The eviction is delivered by the `notify_event` poller, not synchronously, so for a few
+/// seconds after a change a replica can still serve the old address. Handing that to a read or a
+/// job dispatch costs one stale answer. Writing it into a row does not: the row outlives the
+/// eviction, later reads trust it, and nothing re-derives it — so a principal and an address that
+/// name different accounts become permanent. Use this wherever the result is stored, including
+/// the lookups that validate a pair before it is stored.
+pub async fn get_email_from_permissioned_as_uncached<'c>(
+    permissioned_as: &str,
+    workspace_id: &str,
+    db: impl sqlx::PgExecutor<'c>,
+) -> crate::error::Result<String> {
+    get_email_from_permissioned_as_inner(permissioned_as, workspace_id, db, false).await
+}
+
+async fn get_email_from_permissioned_as_inner<'c>(
+    permissioned_as: &str,
+    workspace_id: &str,
+    db: impl sqlx::PgExecutor<'c>,
+    use_cache: bool,
 ) -> crate::error::Result<String> {
     if let Some(username) = permissioned_as.strip_prefix(PERMISSIONED_AS_USER_PREFIX) {
-        let lookup = EmailCacheKey(workspace_id, username);
-        if let Some((email, cached_at)) = EMAIL_CACHE.get(&lookup) {
-            if cached_at.elapsed().as_secs() < EMAIL_CACHE_TTL_SECS {
-                return Ok(email);
+        if use_cache {
+            let lookup = EmailCacheKey(workspace_id, username);
+            if let Some((email, cached_at)) = EMAIL_CACHE.get(&lookup) {
+                if cached_at.elapsed().as_secs() < EMAIL_CACHE_TTL_SECS {
+                    return Ok(email);
+                }
             }
         }
         let email = resolve_username_to_email(workspace_id, username, db)