From 793c31a54e5259025b6eb25b738a3e447c8f470f Mon Sep 17 00:00:00 2001
From: Alex Godbehere <alex.godbehere@gmail.com>
Date: Thu, 9 Jul 2026 12:23:26 +0100
Subject: [PATCH] Fix unbounded subscription leak in auth ACL tracking

track_targets used mergeMap over the principal stream, which keeps
every inner acl_for subscription alive. track_kerberos re-emits the
principal UUID on every identities refetch (one per identity write),
so long-lived notify sessions accumulated inner subscriptions without
bound. Every subsequent emission then fanned out through all of them,
pegging the event loop and logging thousands of ACL dumps per second;
with the service unresponsive, MQTT logins and the Keycloak federation
SPI timed out cluster-wide.

Use switchMap so re-resolution of the principal replaces the inner
subscription, dedupe the resolved UUID in track_kerberos, and dedupe
the computed target set so downstream combineLatest pipelines only
re-fire when the ACL actually changes.
---
 acs-auth/lib/dataflow.js | 16 ++++++++++++++--
 1 file changed, 14 insertions(+), 2 deletions(-)

diff --git a/acs-auth/lib/dataflow.js b/acs-auth/lib/dataflow.js
index aed289138..1ed23672d 100644
--- a/acs-auth/lib/dataflow.js
+++ b/acs-auth/lib/dataflow.js
@@ -200,13 +200,20 @@ export class DataFlow {
         const principal$ = valid_uuid(input)
             ? rx.of(input)
             : this.track_kerberos(input);
+        /* switchMap, not mergeMap: mergeMap keeps every inner
+         * subscription alive, so each re-emission of the principal
+         * (one per identity write, via the identities refetch) added
+         * another permanent subscription to acl_for. Long-lived
+         * notify sessions accumulated these without bound and every
+         * subsequent emission fanned out through all of them. */
         return rxx.rx(
             principal$,
-            rx.mergeMap(p => p ? this.acl_for(p) : rx.of([])),
+            rx.switchMap(p => p ? this.acl_for(p) : rx.of([])),
             rx.map(acl => imm.Seq(acl)
                 .filter(e => e.permission == perm)
                 .map(e => e.target)
                 .toSet()),
+            rx.distinctUntilChanged(imm.is),
             rx.tap(ptd =>
                 this.log("Permitted %s for %s: %o", perm, input, ptd.toJS())));
     }
@@ -288,10 +295,15 @@ export class DataFlow {
     }
 
     track_kerberos (upn) {
+        /* The identities seq refetches the whole table on every
+         * identity write, so it re-emits equal content freely. Dedupe
+         * on the resolved UUID so downstream pipelines only see a
+         * change when the mapping actually changes. */
         return rxx.rx(
             this.identities,
             rx.map(ids => ids.filter(i => i.kind == "kerberos" && i.name == upn)),
-            rx.map(acc => acc[0]?.uuid));
+            rx.map(acc => acc[0]?.uuid),
+            rx.distinctUntilChanged());
     }
 
     find_kerberos (upn) {