test: pin JWKS single-flight with a concurrent cold burst

A stub issuer that counts inbound connections, ten concurrent cached_jwks calls,
and an assertion that exactly one fetch is made. Pins the property the caching
layer exists for, which no test covered before.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VF3v6LA9399gNphmZaHYG3

backend/windmill-common/src/guest_jwt.rs

@@ -469,4 +469,49 @@ mod tests {
     fn a_non_key_pem_is_rejected() {
         assert!(decoding_key_from_pem("-----BEGIN CERTIFICATE-----\nnope\n-----END CERTIFICATE-----").is_err());
     }
+
+    #[tokio::test]
+    async fn a_concurrent_cold_burst_makes_one_jwks_fetch() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        use tokio::io::AsyncWriteExt;
+        // The stub listens on loopback, which SSRF validation refuses without this.
+        unsafe { std::env::set_var("ALLOW_PRIVATE_GUEST_JWKS_URLS", "true") };
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let body = format!(
+            r#"{{"keys":[{{"kty":"EC","crv":"P-256","kid":"k1","x":"{PUB1_X}","y":"{PUB1_Y}"}}]}}"#
+        );
+        let hits = std::sync::Arc::new(AtomicUsize::new(0));
+        let hits_srv = hits.clone();
+        tokio::spawn(async move {
+            loop {
+                let (mut sock, _) = listener.accept().await.unwrap();
+                hits_srv.fetch_add(1, Ordering::SeqCst);
+                let resp = format!(
+                    "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
+                    body.len(),
+                    body
+                );
+                // Delay so the other callers pile onto the single-flight lock before the
+                // leader's fetch returns.
+                tokio::time::sleep(Duration::from_millis(150)).await;
+                let _ = sock.write_all(resp.as_bytes()).await;
+                let _ = sock.shutdown().await;
+            }
+        });
+        let url = format!("http://127.0.0.1:{}/jwks.json", addr.port());
+        let mut handles = Vec::new();
+        for _ in 0..10 {
+            let u = url.clone();
+            handles.push(tokio::spawn(async move { cached_jwks(&u).await.map(|e| e.keys.len()) }));
+        }
+        for h in handles {
+            assert_eq!(h.await.unwrap().unwrap(), 1, "each caller resolves the one key");
+        }
+        assert_eq!(
+            hits.load(Ordering::SeqCst),
+            1,
+            "single-flight: a concurrent cold burst makes one fetch"
+        );
+    }
 }