Live-subscription re-auth sweep and admitted-target retention scale per subscription

[harper-joseph]

Problem

Two per-subscription costs landed with continuous re-authorization (#1414) and the operation-scoped read authorization restore (#1915). Under an MQTT fan-out workload (~150k+ subscriptions per worker; production Fabric nodes measured at 157,894 subs on one worker) they are significant:

1. The 30s re-auth sweep scales per subscription. server/liveSubscriptionAuth.ts runs findAndValidateUser + a deep cloneRequestTarget + allowRead serially for every registered subscription, every 30 seconds — even though a fan-out workload collapses to a handful of distinct (resource, user, target) admissions (29 distinct topics in the measured production case). At 150k+ subs/worker that is an estimated 0.5–1.5 cores of steady-state CPU across a 16-worker node, plus allocation churn feeding GC (measured at 26.4% of worker CPU in a production profile).
2. admittedTarget is retained per subscription. Since Restore operation-scoped read authorization and add explicit row filters #1915 (cc8ebfb), every subscribe deep-clones the RequestTarget and the recheck closure retains it for the life of the subscription. Measured in a local load repro (1,000 WS-MQTT conns × 15 subs): RequestTarget count doubled 15,000 → 30,000, ≈ +0.4 KB/sub (~6 KB/conn, ~7% of subscription heap).

Proposed fix (implemented and verified; full patch attached in the first comment)

Group registrations by a canonical admission identity — resource class (WeakMap'd id) + username + canonical target form. The sweep then runs one findAndValidateUser per username and one allowRead recheck per distinct admission; per-subscription work drops to a sync token-expiry comparison. Duplicate admitted-target clones are discarded at registration, so at most one snapshot is retained per distinct admission. Targets carrying reference-identity values (rowFilter functions, class instances, circular refs) have no safe canonical form and register ungrouped — preserving today's per-subscription behavior for them. Per-entry contexts still get context.user advanced to the fresh user before the group recheck runs.

The patch also includes two small delivery-loop items in server/DurableSubscriptionsSession.ts: yield every 100 delivered messages instead of allocating a promise + macrotask hop per message (3.5× loop overhead in fan-out benchmarks; real backpressure is the awaited listener/socket-drain, the yield is only event-loop fairness), and a shared frozen empty checkPermission object.

Verification (local load repro: 1,000 WS-MQTT connections × 15 subscriptions, 29 topics, QoS 1)

• RequestTarget 30,000 → 15,029 (the 29 = one snapshot per distinct admission)
• 5.01 → 4.79 KB/sub worker heap
• sweep at a 3s test interval: ~0.1% of worker CPU (20 / 17,771 profile samples); per-sweep work goes from 15,000 × (validate + clone + allowRead) to 1 validate + 29 rechecks
• 24 new unit tests (grouping, fail-closed on validation/recheck errors, per-entry expiry within a group, teardown/unregister, key canonicalization + separator-collision resistance) and the full integrationTests/security/subscription-revocation.test.ts suite (6/6: drop_user, alter_role, token expiry × WS/MQTT) pass; oxlint + prettier clean.

Note for whoever picks this up: a cross-model review (per engineering guidelines) has not been run on the patch yet.

---

Filed by an AI agent (Claude Code) from a subscription-path memory/CPU investigation; numbers from a production 5.1.26 Fabric heap snapshot + CPU profile and a local 15k-subscription load repro against this repo.

[harper-joseph]

Full implemented patch (applies to main @ 24f67d8). Attached so the work branch can be deleted; re-apply with git am.

subscription-overhead.patch (~35 KB)

From b472f533c27c1739fbafc4d126059ab82f7ac28c Mon Sep 17 00:00:00 2001
From: Joe <joseph@harperdb.io>
Date: Tue, 4 Aug 2026 18:01:13 -0400
Subject: [PATCH] perf(subscriptions): dedupe live-subscription re-auth by
 admission; batch delivery-loop yields
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

liveSubscriptionAuth now groups registrations by (resource class, user,
canonical target): one allowRead recheck per distinct admission per sweep
and one findAndValidateUser per username, instead of one of each per
subscription (~150k+/worker under MQTT fan-out). Grouping also drops the
per-subscription admittedTarget clone (#1915 doubled RequestTarget per
subscription) — at most one snapshot is retained per distinct admission.
Targets with reference-identity values (rowFilter etc.) register ungrouped,
preserving per-subscription behavior.

DurableSubscriptionsSession: yield the delivery loop every 100 messages
instead of allocating a promise + macrotask hop per delivered message
(3.5x loop overhead in fan-out benchmarks), and share a frozen empty
checkPermission object.

WIP branch — cross-model review still pending before any PR.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 resources/Resource.ts                         | 117 +++++++--
 server/DurableSubscriptionsSession.ts         |  18 +-
 server/liveSubscriptionAuth.ts                | 193 ++++++++++++---
 .../resources/liveSubscriptionAuthKey.test.js | 106 ++++++++
 unitTests/server/liveSubscriptionAuth.test.js | 228 ++++++++++++++++++
 5 files changed, 613 insertions(+), 49 deletions(-)
 create mode 100644 unitTests/resources/liveSubscriptionAuthKey.test.js
 create mode 100644 unitTests/server/liveSubscriptionAuth.test.js

diff --git a/resources/Resource.ts b/resources/Resource.ts
index 22d122c94..714c8cb04 100644
--- a/resources/Resource.ts
+++ b/resources/Resource.ts
@@ -817,7 +817,11 @@ function transactional(
 			const runAction = (data: any) => {
 				// Capture the complete target after the initial allowRead has narrowed it, but before
 				// subscribe/connect implementations can mutate it. Every later recheck gets a fresh clone.
+				// The authKey (same pre-mutation snapshot) lets identical admissions share one registry
+				// group — when a group already exists the clone is discarded at registration, so at most
+				// one snapshot is retained per distinct admission rather than one per subscription.
 				const admittedTarget = isSubscribeAction ? cloneRequestTarget(query) : undefined;
+				const admittedKey = isSubscribeAction ? liveSubscriptionAuthKey(resource, context?.user?.username, query) : null;
 				// getResource creates a fresh receiver for this dispatch. Marking that receiver, rather
 				// than caller-owned target identity, preserves the static (target, message) signature when
 				// an override copies the target or delegates asynchronously after its own return settles.
@@ -825,7 +829,7 @@ function transactional(
 				const result = action(resource, query, context, data);
 				if (!isSubscribeAction) return result;
 				return when(result, (subscription: any) => {
-					registerLiveSubscriptionForContext(subscription, resource, admittedTarget, context);
+					registerLiveSubscriptionForContext(subscription, resource, admittedTarget, admittedKey, context);
 					return subscription;
 				});
 			};
@@ -942,7 +946,13 @@ function hasPermissionControl(value: any, seen = new WeakSet<object>()): boolean
 	return false;
 }
 
-function registerLiveSubscriptionForContext(subscription: any, resource: any, admittedTarget: any, context: Context) {
+function registerLiveSubscriptionForContext(
+	subscription: any,
+	resource: any,
+	admittedTarget: any,
+	admittedKey: string | null,
+	context: Context
+) {
 	const user: any = context?.user;
 	const username = user?.username;
 	// Internal watchers, replication and local-bypass have no user principal — nothing to re-authorize.
@@ -952,24 +962,103 @@ function registerLiveSubscriptionForContext(subscription: any, resource: any, ad
 		username,
 		// JWT exp of the bearer credential (set by the auth layer); undefined for password/mTLS/session.
 		authExpiresAt: user.authExpiresAt,
-		recheck: async () => {
-			// Re-fetch current user state — the user/role cache is rebuilt on mutations — so a dropped or
-			// role-stripped user no longer authorizes.
-			const { findAndValidateUser } = require('../security/user');
-			const fresh: any = await findAndValidateUser(username, undefined, false);
-			if (!fresh?.role) return false;
-			// Advance the subscription's context to the fresh user so downstream checks — context.user
-			// and getCurrentUser() (which reads the resource's context) — evaluate against current state,
-			// not the stale user captured at subscribe time.
-			if (context) (context as any).user = fresh;
-			// Re-run the same operation-level allowRead that granted the subscription.
+		context,
+		authKey: admittedKey,
+		// The registry validates the user once per sweep (shared across the user's groups) and passes
+		// the fresh user in; it also advances each member's context.user beforehand. This closure only
+		// re-runs the same operation-level allowRead that granted the subscription.
+		recheck: async (fresh: any, entryContext: any) => {
 			const reTarget: any = cloneRequestTarget(admittedTarget);
 			reTarget.checkPermission = fresh.role?.permission;
-			return !!(await resource.allowRead(fresh, reTarget, context));
+			return !!(await resource.allowRead(fresh, reTarget, entryContext));
 		},
 	});
 }
 
+// Class identity component of the live-subscription authKey — a WeakMap'd counter rather than the
+// class name, since two different resource classes can share a name.
+const liveAuthClassIds = new WeakMap<Function, number>();
+let nextLiveAuthClassId = 0;
+
+/**
+ * Stable identity of a subscription admission — (resource class, username, canonical target) — used
+ * to share one re-authorization group among identical admissions (see liveSubscriptionAuth.ts).
+ * Returns null when the target has no safe canonical form (functions, class instances, circular
+ * references), in which case the subscription registers ungrouped.
+ */
+function liveSubscriptionAuthKey(resource: any, username: string | undefined, query: any): string | null {
+	if (!username) return null;
+	const resourceClass = resource?.constructor;
+	if (typeof resourceClass !== 'function') return null;
+	const targetKey = canonicalTargetKey(query);
+	if (targetKey == null) return null;
+	let classId = liveAuthClassIds.get(resourceClass);
+	if (classId === undefined) liveAuthClassIds.set(resourceClass, (classId = ++nextLiveAuthClassId));
+	// JSON-encode the username so a name containing the separator can't shift segment boundaries
+	// and collide with another (username, target) pair.
+	return `${classId}|${JSON.stringify(username)}|${targetKey}`;
+}
+
+/**
+ * Canonical string form of a request target, covering exactly what cloneRequestTarget snapshots:
+ * the URLSearchParams state plus own enumerable properties (checkPermission excluded). Returns null
+ * for values the clone would carry by reference (functions, class instances) or circular structures
+ * — identity matters for those, so such targets must not share a group.
+ */
+function canonicalTargetKey(source: any): string | null {
+	if (source == null || typeof source !== 'object') return null;
+	const seen = new WeakSet<object>();
+	const canonicalValue = (value: any): string | null => {
+		if (value === null) return 'null';
+		switch (typeof value) {
+			case 'string':
+				return JSON.stringify(value);
+			case 'number':
+			case 'boolean':
+			case 'bigint':
+				return String(value);
+			case 'undefined':
+				return 'undefined';
+			case 'object':
+				break;
+			default:
+				return null; // function or symbol — reference identity, not canonicalizable
+		}
+		if (seen.has(value)) return null; // circular
+		seen.add(value);
+		if (Array.isArray(value)) {
+			const parts: string[] = [];
+			for (const element of value) {
+				const part = canonicalValue(element);
+				if (part == null) return null;
+				parts.push(part);
+			}
+			return `[${parts.join(',')}]`;
+		}
+		const prototype = Object.getPrototypeOf(value);
+		if (prototype !== Object.prototype && prototype !== null) return null; // class instance
+		return canonicalProperties(value);
+	};
+	const canonicalProperties = (value: object): string | null => {
+		const parts: string[] = [];
+		for (const key of Object.keys(value).sort()) {
+			if (key === 'checkPermission') continue; // stripped by cloneRequestTarget; role-dependent
+			const part = canonicalValue((value as any)[key]);
+			if (part == null) return null;
+			parts.push(`${JSON.stringify(key)}:${part}`);
+		}
+		return `{${parts.join(',')}}`;
+	};
+	seen.add(source);
+	const ownProperties = canonicalProperties(source);
+	if (ownProperties == null) return null;
+	const search = source instanceof URLSearchParams ? source.toString() : '';
+	return `${search}|${ownProperties}`;
+}
+
+// Test-only aliases (unitTests/resources/liveSubscriptionAuthKey.test.js)
+export { liveSubscriptionAuthKey as _liveSubscriptionAuthKeyForTest, canonicalTargetKey as _canonicalTargetKeyForTest };
+
 const KNOWN_METHODS = ['get', 'head', 'put', 'post', 'delete', 'patch', 'query', 'move', 'copy'];
 type ClientErrorWithMethods = ClientError & { allow: string[]; method: string };
 export function missingMethod(resource: any, method: string) {
diff --git a/server/DurableSubscriptionsSession.ts b/server/DurableSubscriptionsSession.ts
index 16142860f..ba8a8ac01 100644
--- a/server/DurableSubscriptionsSession.ts
+++ b/server/DurableSubscriptionsSession.ts
@@ -10,6 +10,14 @@ import { RequestTarget } from '../resources/RequestTarget';
 import { cloneDeep } from 'lodash';
 
 const AWAITING_ACKS_HIGH_WATER_MARK = 100;
+// Deliveries between event-loop yields in a subscription's delivery loop. Yielding per message
+// costs a promise allocation and a macrotask hop per delivered message (~3.5x the loop's own
+// overhead in fan-out benchmarks); yielding every N keeps event-loop fairness without that tax.
+// Matches Table.ts's REPLAY_YIELD_INTERVAL batch size.
+const DELIVERY_YIELD_INTERVAL = 100;
+// Shared frozen fallback for users with no role permissions — one subscription-lifetime object
+// per session's addSubscription call otherwise.
+const NO_PERMISSIONS = Object.freeze({});
 let _DurableSession: any;
 function getDurableSession() {
 	if (!_DurableSession) {
@@ -262,7 +270,7 @@ class SubscriptionsSession {
 			onlyChildren,
 			startTime,
 			omitCurrent,
-			checkPermission: this.user?.role?.permission ?? {},
+			checkPermission: this.user?.role?.permission ?? NO_PERMISSIONS,
 		});
 		const resourcePath = entry.path;
 		const resource = entry.Resource;
@@ -278,6 +286,7 @@ class SubscriptionsSession {
 			if (!subscription[Symbol.asyncIterator])
 				throw new Error(`Subscription is not (async) iterable for topic ${topic}`);
 			const _result = (async () => {
+				let deliveriesSinceYield = 0;
 				for await (const update of subscription) {
 					try {
 						if (!update || typeof update !== 'object') continue;
@@ -310,7 +319,12 @@ class SubscriptionsSession {
 							await new Promise((resolve) =>
 								setTimeout(resolve, this.awaitingAcks.size - AWAITING_ACKS_HIGH_WATER_MARK)
 							);
-						} else await new Promise(setImmediate); // yield event turn
+						} else if (++deliveriesSinceYield >= DELIVERY_YIELD_INTERVAL) {
+							// yield the event turn periodically; real backpressure comes from awaiting the
+							// listener (socket drain) above, this only preserves event-loop fairness
+							deliveriesSinceYield = 0;
+							await new Promise(setImmediate);
+						}
 					} catch (error) {
 						warn(error);
 					}
diff --git a/server/liveSubscriptionAuth.ts b/server/liveSubscriptionAuth.ts
index d81ca0233..da00768a8 100644
--- a/server/liveSubscriptionAuth.ts
+++ b/server/liveSubscriptionAuth.ts
@@ -12,25 +12,56 @@ import hdbLogger from '../utility/logging/harper_logger.ts';
  * Triggers: (1) immediately on the ITC user-change broadcast (the same signal that rebuilds the
  * user/role cache), and (2) on a fixed interval as a backstop and to catch token expiry, which is
  * not event-signaled.
+ *
+ * Scale: an MQTT fan-out workload holds ~150k+ subscriptions per worker, but they collapse to a
+ * handful of distinct (resource, user, target) admissions — so subscriptions register under an
+ * `authKey` and identical admissions share one group. The sweep then runs one user validation per
+ * username and one recheck per group, instead of one of each per subscription, and only the cheap
+ * token-expiry comparison stays per-subscription. Registrations without a safe identity
+ * (authKey == null) get a private group each, which degrades to the per-subscription behavior.
  */
 
 // Backstop interval; also catches token expiry, which is not event-signaled. Overridable for tests.
 const RECHECK_INTERVAL_MS = Number(process.env.HARPER_SUBSCRIPTION_REAUTH_INTERVAL_MS) || 30_000;
 
-interface LiveSubscription {
-	username: string;
+interface LiveSubscriptionEntry {
 	/** JWT `exp` (seconds since epoch) of the credential the subscription was opened with, if any. */
 	authExpiresAt?: number;
-	/** Returns true if the principal is still authorized for this subscription. */
-	recheck: () => Promise<boolean>;
+	/** The subscription's own context; its `user` is advanced to the freshly validated user each sweep. */
+	context?: any;
+	group: RecheckGroup;
 	/** Stop delivery and tear down the subscription. */
 	terminate: () => void;
 }
 
-const registry = new Set<LiveSubscription>();
+interface RecheckGroup {
+	username: string;
+	/** Present for deduplicated groups; undefined for private (non-deduplicable) groups. */
+	key?: string;
+	/**
+	 * Re-runs the admission-time authorization against a freshly validated user. Shared by every
+	 * subscription in the group — they were admitted with an identical (resource, user, target).
+	 */
+	recheck: (freshUser: any, context: any) => Promise<boolean>;
+	entries: Set<LiveSubscriptionEntry>;
+}
+
+const groupsByKey = new Map<string, RecheckGroup>();
+const groupsByUser = new Map<string, Set<RecheckGroup>>();
+let entryCount = 0;
 let sweepTimer: any = null;
 let itcListenerInstalled = false;
 let sweeping = false;
+// Test seam: sweep() validates users through this indirection so unit tests can stub it.
+let findUser: ((username: string) => Promise<any>) | null = null;
+
+function validateUser(username: string): Promise<any> {
+	if (!findUser) {
+		const { findAndValidateUser } = require('../security/user');
+		findUser = (name: string) => findAndValidateUser(name, undefined, false);
+	}
+	return findUser(username);
+}
 
 function ensureStarted(): void {
 	if (!sweepTimer) {
@@ -54,29 +85,64 @@ function ensureStarted(): void {
 }
 
 function stopIfIdle(): void {
-	if (registry.size === 0 && sweepTimer) {
+	if (entryCount === 0 && sweepTimer) {
 		clearInterval(sweepTimer);
 		sweepTimer = null;
 	}
 }
 
+function removeEntry(entry: LiveSubscriptionEntry): void {
+	const group = entry.group;
+	if (!group.entries.delete(entry)) return; // already removed (end() and 'close' both fire)
+	entryCount--;
+	if (group.entries.size === 0) {
+		if (group.key !== undefined) groupsByKey.delete(group.key);
+		const userGroups = groupsByUser.get(group.username);
+		if (userGroups) {
+			userGroups.delete(group);
+			if (userGroups.size === 0) groupsByUser.delete(group.username);
+		}
+	}
+	stopIfIdle();
+}
+
 /**
  * Register a live subscription for continuous re-authorization. The returned subscription's normal
  * teardown (end()/close) automatically unregisters it, so callers don't need to.
+ *
+ * `authKey` is a stable identity of the admission (resource class + username + canonical target);
+ * registrations sharing an authKey share one group — and one recheck per sweep. Pass null when the
+ * target has no safe canonical form (e.g. it carries functions); such registrations are never
+ * shared. When an authKey matches an existing group, the caller's `recheck` (and everything it
+ * retains, like the admitted-target snapshot) is discarded in favor of the group's.
  */
 export function registerLiveSubscription(opts: {
 	subscription: any;
 	username: string;
 	authExpiresAt?: number;
-	recheck: () => Promise<boolean>;
+	context?: any;
+	authKey?: string | null;
+	recheck: (freshUser: any, context: any) => Promise<boolean>;
 }): void {
-	const { subscription, username, authExpiresAt, recheck } = opts;
+	const { subscription, username, authExpiresAt, context, authKey, recheck } = opts;
 	if (!subscription || typeof subscription !== 'object' || subscription.closed) return;
 
-	const entry: LiveSubscription = {
-		username,
+	let group = authKey != null ? groupsByKey.get(authKey) : undefined;
+	if (!group) {
+		group = { username, recheck, entries: new Set() };
+		if (authKey != null) {
+			group.key = authKey;
+			groupsByKey.set(authKey, group);
+		}
+		let userGroups = groupsByUser.get(username);
+		if (!userGroups) groupsByUser.set(username, (userGroups = new Set()));
+		userGroups.add(group);
+	}
+
+	const entry: LiveSubscriptionEntry = {
 		authExpiresAt,
-		recheck,
+		context,
+		group,
 		terminate: () => {
 			// end() removes the subscription from the broadcast loop and closes its iterable queue.
 			try {
@@ -88,12 +154,10 @@ export function registerLiveSubscription(opts: {
 			}
 		},
 	};
-	registry.add(entry);
+	group.entries.add(entry);
+	entryCount++;
 
-	const unregister = () => {
-		registry.delete(entry);
-		stopIfIdle();
-	};
+	const unregister = () => removeEntry(entry);
 	// Both transports ultimately call end() on normal teardown (MQTT unsubscribe/disconnect; SSE close
 	// is wired to end()); wrap it so a closed stream never leaks a registry entry. Also listen for
 	// 'close' to cover any iterable that closes without an end().
@@ -109,32 +173,66 @@ export function registerLiveSubscription(opts: {
 	ensureStarted();
 }
 
+function revokeEntry(entry: LiveSubscriptionEntry, reason: string): void {
+	removeEntry(entry);
+	try {
+		entry.terminate();
+	} catch {
+		/* ignore */
+	}
+	hdbLogger.trace?.(`liveSubscriptionAuth: revoked subscription for ${entry.group.username} (${reason})`);
+}
+
 async function sweep(): Promise<void> {
 	if (sweeping) return; // a slow recheck must not overlap with the next tick/event
 	sweeping = true;
 	try {
 		const now = Date.now();
-		for (const entry of registry) {
+		for (const [username, userGroups] of [...groupsByUser]) {
+			// One user validation per username per sweep, shared by all of the user's groups.
+			let fresh: any = null;
+			let lookupError: Error | null = null;
 			try {
-				const expired = entry.authExpiresAt != null && now >= entry.authExpiresAt * 1000;
-				const stillAuthorized = expired ? false : await entry.recheck();
-				const revoke = expired || !stillAuthorized;
-				if (revoke) {
-					registry.delete(entry);
-					entry.terminate();
-				}
+				fresh = await validateUser(username);
 			} catch (error) {
-				// fail closed: if authorization can't be confirmed, revoke
-				registry.delete(entry);
-				try {
-					entry.terminate();
-				} catch {
-					/* ignore */
-				}
+				lookupError = error as Error;
+			}
+			const userValid = !lookupError && fresh?.role;
+			if (lookupError) {
 				hdbLogger.warn?.(
-					`liveSubscriptionAuth: revoked subscription for ${entry.username} after recheck error: ${(error as Error).message}`
+					`liveSubscriptionAuth: revoking subscriptions for ${username} after validation error: ${lookupError.message}`
 				);
 			}
+			for (const group of [...userGroups]) {
+				if (group.entries.size === 0) continue; // emptied by an earlier terminate() cascade
+				if (!userValid) {
+					// fail closed: dropped user, stripped role, or a validation error revokes everything
+					for (const entry of [...group.entries]) revokeEntry(entry, 'user no longer valid');
+					continue;
+				}
+				// Advance each subscription's context to the fresh user BEFORE the recheck — so the
+				// recheck itself and downstream checks (context.user, getCurrentUser()) evaluate against
+				// current state, not the stale user captured at subscribe time.
+				for (const entry of group.entries) {
+					if (entry.context) entry.context.user = fresh;
+				}
+				let authorized = false;
+				try {
+					// Any live member's context serves the group-level check; the admission was identical.
+					const entryContext = group.entries.values().next().value?.context;
+					authorized = !!(await group.recheck(fresh, entryContext));
+				} catch (error) {
+					// fail closed: if authorization can't be confirmed, revoke
+					hdbLogger.warn?.(
+						`liveSubscriptionAuth: revoking subscriptions for ${username} after recheck error: ${(error as Error).message}`
+					);
+				}
+				for (const entry of [...group.entries]) {
+					const expired = entry.authExpiresAt != null && now >= entry.authExpiresAt * 1000;
+					if (!authorized) revokeEntry(entry, 'no longer authorized');
+					else if (expired) revokeEntry(entry, 'credential expired');
+				}
+			}
 		}
 	} finally {
 		sweeping = false;
@@ -144,5 +242,34 @@ async function sweep(): Promise<void> {
 
 /** Test-only: current number of tracked subscriptions. */
 export function _liveSubscriptionCount(): number {
-	return registry.size;
+	return entryCount;
+}
+
+/** Test-only: current number of recheck groups. */
+export function _liveSubscriptionGroupCount(): number {
+	let count = 0;
+	for (const userGroups of groupsByUser.values()) count += userGroups.size;
+	return count;
+}
+
+/** Test-only: run one sweep to completion. */
+export function _sweepForTest(): Promise<void> {
+	return sweep();
+}
+
+/** Test-only: stub the user validation the sweep performs. Pass null to restore the default. */
+export function _setUserValidatorForTest(validator: ((username: string) => Promise<any>) | null): void {
+	findUser = validator;
+}
+
+/** Test-only: drop all registrations and stop the sweep timer. */
+export function _resetForTest(): void {
+	groupsByKey.clear();
+	groupsByUser.clear();
+	entryCount = 0;
+	if (sweepTimer) {
+		clearInterval(sweepTimer);
+		sweepTimer = null;
+	}
+	sweeping = false;
 }
diff --git a/unitTests/resources/liveSubscriptionAuthKey.test.js b/unitTests/resources/liveSubscriptionAuthKey.test.js
new file mode 100644
index 000000000..d225c3dea
--- /dev/null
+++ b/unitTests/resources/liveSubscriptionAuthKey.test.js
@@ -0,0 +1,106 @@
+'use strict';
+
+const testUtils = require('../testUtils.js');
+testUtils.preTestPrep();
+
+const assert = require('node:assert');
+const {
+	_liveSubscriptionAuthKeyForTest: authKey,
+	_canonicalTargetKeyForTest: canonicalKey,
+} = require('#src/resources/Resource');
+const { RequestTarget } = require('#src/resources/RequestTarget');
+
+class ResourceA {}
+class ResourceB {}
+const resourceA = new ResourceA();
+const resourceB = new ResourceB();
+
+function target(url, props = {}) {
+	const requestTarget = new RequestTarget(url);
+	Object.assign(requestTarget, props);
+	return requestTarget;
+}
+
+describe('liveSubscriptionAuthKey', () => {
+	it('is stable for identical admissions', () => {
+		const a = authKey(resourceA, 'u1', target('/Msg/topic1', { isCollection: false, startTime: 5 }));
+		const b = authKey(resourceA, 'u1', target('/Msg/topic1', { isCollection: false, startTime: 5 }));
+		assert.ok(a != null);
+		assert.equal(a, b);
+	});
+
+	it('is insensitive to property assignment order', () => {
+		const a = authKey(resourceA, 'u1', target('/Msg/t', { startTime: 5, omitCurrent: true }));
+		const b = authKey(resourceA, 'u1', target('/Msg/t', { omitCurrent: true, startTime: 5 }));
+		assert.equal(a, b);
+	});
+
+	it('discriminates by target path, query string, user, and resource class', () => {
+		const base = () => target('/Msg/topic1');
+		const key = authKey(resourceA, 'u1', base());
+		assert.notEqual(authKey(resourceA, 'u1', target('/Msg/topic2')), key);
+		assert.notEqual(authKey(resourceA, 'u1', target('/Msg/topic1?limit=5')), key);
+		assert.notEqual(authKey(resourceA, 'u2', base()), key);
+		assert.notEqual(authKey(resourceB, 'u1', base()), key);
+	});
+
+	it('discriminates by assigned scalar and structured properties', () => {
+		const key = authKey(resourceA, 'u1', target('/Msg/t', { startTime: 5 }));
+		assert.notEqual(authKey(resourceA, 'u1', target('/Msg/t', { startTime: 6 })), key);
+		assert.notEqual(authKey(resourceA, 'u1', target('/Msg/t', { startTime: 5, onlyChildren: true })), key);
+		const withConditions = (value) =>
+			authKey(resourceA, 'u1', target('/Msg/t', { conditions: [{ attribute: 'a', value }] }));
+		assert.equal(withConditions(1), withConditions(1));
+		assert.notEqual(withConditions(1), withConditions(2));
+	});
+
+	it('ignores checkPermission — it is role-derived, not part of the admission identity', () => {
+		const a = authKey(resourceA, 'u1', target('/Msg/t', { checkPermission: { super_user: true } }));
+		const b = authKey(resourceA, 'u1', target('/Msg/t', { checkPermission: { tables: {} } }));
+		assert.equal(a, b);
+	});
+
+	it('a username containing the separator cannot collide across segment boundaries', () => {
+		const a = authKey(resourceA, 'u1|x', target('/Msg/t'));
+		const b = authKey(resourceA, 'u1', target('x|/Msg/t'));
+		assert.notEqual(a, b);
+	});
+
+	it('returns null without a username', () => {
+		assert.equal(authKey(resourceA, undefined, target('/Msg/t')), null);
+		assert.equal(authKey(resourceA, '', target('/Msg/t')), null);
+	});
+
+	it('returns null for targets carrying reference-identity values', () => {
+		assert.equal(authKey(resourceA, 'u1', target('/Msg/t', { rowFilter: () => true })), null);
+		assert.equal(authKey(resourceA, 'u1', target('/Msg/t', { custom: new Date() })), null);
+		const circular = {};
+		circular.self = circular;
+		assert.equal(authKey(resourceA, 'u1', target('/Msg/t', { circular })), null);
+	});
+});
+
+describe('canonicalTargetKey', () => {
+	it('canonicalizes primitives, arrays, and nested plain objects', () => {
+		const a = canonicalKey(target('/Msg/t', { id: ['a', 1], nested: { x: null, y: undefined, z: false } }));
+		const b = canonicalKey(target('/Msg/t', { nested: { z: false, y: undefined, x: null }, id: ['a', 1] }));
+		assert.ok(a != null);
+		assert.equal(a, b);
+	});
+
+	it('returns null for non-object input', () => {
+		assert.equal(canonicalKey(null), null);
+		assert.equal(canonicalKey('string'), null);
+	});
+
+	it('distinguishes value types that stringify alike', () => {
+		assert.notEqual(
+			canonicalKey(target('/Msg/t', { value: 1 })),
+			canonicalKey(target('/Msg/t', { value: '1' }))
+		);
+		assert.notEqual(
+			canonicalKey(target('/Msg/t', { value: null })),
+			canonicalKey(target('/Msg/t', { value: 'null' }))
+		);
+	});
+});
diff --git a/unitTests/server/liveSubscriptionAuth.test.js b/unitTests/server/liveSubscriptionAuth.test.js
new file mode 100644
index 000000000..9d93db9e1
--- /dev/null
+++ b/unitTests/server/liveSubscriptionAuth.test.js
@@ -0,0 +1,228 @@
+'use strict';
+
+const testUtils = require('../testUtils.js');
+testUtils.preTestPrep();
+
+const assert = require('node:assert');
+const EventEmitter = require('node:events');
+const {
+	registerLiveSubscription,
+	_liveSubscriptionCount,
+	_liveSubscriptionGroupCount,
+	_sweepForTest,
+	_setUserValidatorForTest,
+	_resetForTest,
+} = require('#src/server/liveSubscriptionAuth');
+
+// The subscription shape the registry interacts with: an EventEmitter with end()/closed,
+// mirroring transactionBroadcast's Subscription (IterableEventQueue).
+function makeSubscription() {
+	const subscription = new EventEmitter();
+	subscription.closed = false;
+	subscription.ended = 0;
+	subscription.end = () => {
+		subscription.ended++;
+		subscription.closed = true;
+		subscription.emit('close');
+	};
+	return subscription;
+}
+
+function makeUser(username, overrides = {}) {
+	return { username, role: { permission: {} }, ...overrides };
+}
+
+describe('liveSubscriptionAuth', () => {
+	let validatedUsernames;
+	let userTable;
+
+	beforeEach(() => {
+		_resetForTest();
+		validatedUsernames = [];
+		userTable = new Map();
+		_setUserValidatorForTest(async (username) => {
+			validatedUsernames.push(username);
+			return userTable.get(username);
+		});
+	});
+
+	afterEach(() => {
+		_resetForTest();
+		_setUserValidatorForTest(null);
+	});
+
+	function register(subscription, options = {}) {
+		registerLiveSubscription({
+			subscription,
+			username: options.username ?? 'u1',
+			authExpiresAt: options.authExpiresAt,
+			context: options.context,
+			authKey: 'authKey' in options ? options.authKey : null,
+			recheck: options.recheck ?? (async () => true),
+		});
+	}
+
+	it('groups registrations that share an authKey and rechecks the group once', async () => {
+		userTable.set('u1', makeUser('u1'));
+		let recheckCalls = 0;
+		const sharedRecheck = async () => {
+			recheckCalls++;
+			return true;
+		};
+		const subscriptions = [makeSubscription(), makeSubscription(), makeSubscription()];
+		for (const subscription of subscriptions) {
+			register(subscription, { authKey: '1 u1 {}', recheck: sharedRecheck });
+		}
+		assert.equal(_liveSubscriptionCount(), 3);
+		assert.equal(_liveSubscriptionGroupCount(), 1);
+
+		await _sweepForTest();
+		assert.equal(recheckCalls, 1);
+		assert.deepEqual(validatedUsernames, ['u1']);
+		assert.ok(subscriptions.every((subscription) => !subscription.closed));
+	});
+
+	it('validates each username once per sweep across that user\'s groups', async () => {
+		userTable.set('u1', makeUser('u1'));
+		userTable.set('u2', makeUser('u2'));
+		register(makeSubscription(), { username: 'u1', authKey: '1 u1 {a}' });
+		register(makeSubscription(), { username: 'u1', authKey: '1 u1 {b}' });
+		register(makeSubscription(), { username: 'u2', authKey: '1 u2 {a}' });
+		assert.equal(_liveSubscriptionGroupCount(), 3);
+
+		await _sweepForTest();
+		assert.deepEqual(validatedUsernames.sort(), ['u1', 'u2']);
+	});
+
+	it('gives null-authKey registrations a private group each', () => {
+		register(makeSubscription(), { authKey: null });
+		register(makeSubscription(), { authKey: null });
+		assert.equal(_liveSubscriptionCount(), 2);
+		assert.equal(_liveSubscriptionGroupCount(), 2);
+	});
+
+	it('terminates every subscription in a group when its recheck fails', async () => {
+		userTable.set('u1', makeUser('u1'));
+		const denied = [makeSubscription(), makeSubscription()];
+		const allowed = makeSubscription();
+		for (const subscription of denied) {
+			register(subscription, { authKey: 'denied', recheck: async () => false });
+		}
+		register(allowed, { authKey: 'allowed', recheck: async () => true });
+
+		await _sweepForTest();
+		assert.ok(denied.every((subscription) => subscription.ended === 1));
+		assert.equal(allowed.ended, 0);
+		assert.equal(_liveSubscriptionCount(), 1);
+		assert.equal(_liveSubscriptionGroupCount(), 1);
+	});
+
+	it('fails closed: a recheck that throws revokes the group', async () => {
+		userTable.set('u1', makeUser('u1'));
+		const subscription = makeSubscription();
+		register(subscription, {
+			authKey: 'throws',
+			recheck: async () => {
+				throw new Error('permission source unavailable');
+			},
+		});
+		await _sweepForTest();
+		assert.equal(subscription.ended, 1);
+		assert.equal(_liveSubscriptionCount(), 0);
+	});
+
+	it('fails closed: a user validation error revokes all of the user\'s subscriptions', async () => {
+		_setUserValidatorForTest(async () => {
+			throw new Error('user store unavailable');
+		});
+		const subscriptions = [makeSubscription(), makeSubscription()];
+		register(subscriptions[0], { authKey: 'a' });
+		register(subscriptions[1], { authKey: 'b' });
+		await _sweepForTest();
+		assert.ok(subscriptions.every((subscription) => subscription.ended === 1));
+		assert.equal(_liveSubscriptionCount(), 0);
+	});
+
+	it('revokes a dropped or role-stripped user without running rechecks', async () => {
+		userTable.set('u1', undefined); // dropped
+		let recheckCalls = 0;
+		const subscription = makeSubscription();
+		register(subscription, {
+			authKey: 'k',
+			recheck: async () => {
+				recheckCalls++;
+				return true;
+			},
+		});
+		await _sweepForTest();
+		assert.equal(subscription.ended, 1);
+		assert.equal(recheckCalls, 0);
+	});
+
+	it('expires only the entries whose credential lapsed, keeping groupmates alive', async () => {
+		userTable.set('u1', makeUser('u1'));
+		const expired = makeSubscription();
+		const current = makeSubscription();
+		const past = Date.now() / 1000 - 10;
+		const future = Date.now() / 1000 + 3600;
+		register(expired, { authKey: 'shared', authExpiresAt: past });
+		register(current, { authKey: 'shared', authExpiresAt: future });
+		assert.equal(_liveSubscriptionGroupCount(), 1);
+
+		await _sweepForTest();
+		assert.equal(expired.ended, 1);
+		assert.equal(current.ended, 0);
+		assert.equal(_liveSubscriptionCount(), 1);
+	});
+
+	it('advances every group member\'s context.user to the freshly validated user', async () => {
+		const fresh = makeUser('u1');
+		userTable.set('u1', fresh);
+		const contexts = [{ user: 'stale' }, { user: 'stale' }];
+		register(makeSubscription(), { authKey: 'k', context: contexts[0] });
+		register(makeSubscription(), { authKey: 'k', context: contexts[1] });
+
+		await _sweepForTest();
+		assert.ok(contexts.every((context) => context.user === fresh));
+	});
+
+	it('unregisters on subscription.end() and drops the emptied group', () => {
+		const subscriptions = [makeSubscription(), makeSubscription()];
+		register(subscriptions[0], { authKey: 'k' });
+		register(subscriptions[1], { authKey: 'k' });
+		assert.equal(_liveSubscriptionCount(), 2);
+
+		subscriptions[0].end();
+		assert.equal(_liveSubscriptionCount(), 1);
+		assert.equal(_liveSubscriptionGroupCount(), 1);
+
+		subscriptions[1].end();
+		assert.equal(_liveSubscriptionCount(), 0);
+		assert.equal(_liveSubscriptionGroupCount(), 0);
+	});
+
+	it('unregisters on a bare close event (iterables that close without end())', () => {
+		const subscription = makeSubscription();
+		register(subscription, { authKey: 'k' });
+		subscription.emit('close');
+		assert.equal(_liveSubscriptionCount(), 0);
+	});
+
+	it('never registers an already-closed subscription', () => {
+		const subscription = makeSubscription();
+		subscription.closed = true;
+		register(subscription, { authKey: 'k' });
+		assert.equal(_liveSubscriptionCount(), 0);
+	});
+
+	it('a new registration after a group emptied starts a fresh group under the same key', async () => {
+		userTable.set('u1', makeUser('u1'));
+		const first = makeSubscription();
+		register(first, { authKey: 'k', recheck: async () => false });
+		first.end();
+		const second = makeSubscription();
+		register(second, { authKey: 'k', recheck: async () => true });
+		await _sweepForTest();
+		assert.equal(second.ended, 0);
+	});
+});
-- 
2.39.5 (Apple Git-154)

Posted by an AI agent (Claude Code).