Disable suspended tenants (#379)

* add authInfo to align with CDS

* upgrade cds

* wip

* wip

* wip

* wip

* wip

* wip

* wip

* wip

* wip

* wip

Author: soccermax

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 ### Added
 
 - Added `authInfo` to cds.User as CDS 9.3 deprecated `tokenInfo`.
+- Disable the automatic processing of suspended tenants. Can be turned off using `disableProcessingOfSuspendedTenants`.
 
 ## v1.10.10 - 2025-07-09
 

docs/setup/index.md

@@ -84,6 +84,7 @@ The table includes the parameter name, a description of its purpose, and the def
 | redisNamespace                       | Prefix applied to all Redis interactions. Use this when multiple microservices share a Redis instance to prevent naming conflicts.                                                                                                                                                                                                                                                                                                          | null           | no                        |
 | redisOptions                         | The option is provided to customize settings when creating Redis clients. The object is spread at the root level for creating a client and within the `default` options for cluster clients.                                                                                                                                                                                                                                                | {}             | no                        |
 | crashOnRedisUnavailable              | If enabled, the application will crash if Redis is unavailable during the connection check.                                                                                                                                                                                                                                                                                                                                                 | false          | false                     |
+| disableProcessingOfSuspendedTenants  | Disables the processing of suspended tenants. A tenant is considered suspended if XSUAA returns `404` for the given tenant.                                                                                                                                                                                                                                                                                                                 | true           | true                      |
 
 # Configure Redis
 

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "@cap-js-community/event-queue",
-  "version": "1.10.11",
+  "version": "1.11.0-beta.4",
   "description": "An event queue that enables secure transactional processing of asynchronous and periodic events, featuring instant event processing with Redis Pub/Sub and load distribution across all application instances.",
   "main": "src/index.js",
   "types": "src/index.d.ts",

package.json

@@ -119,6 +119,7 @@ class Config {
   #configEvents;
   #configPeriodicEvents;
   #enableAdminService;
+  #disableProcessingOfSuspendedTenants;
   static #instance;
   constructor() {
     this.#logger = cds.log(COMPONENT_NAME);
@@ -984,6 +985,14 @@ class Config {
     this.#enableAdminService = value;
   }
 
+  get disableProcessingOfSuspendedTenants() {
+    return this.#disableProcessingOfSuspendedTenants;
+  }
+
+  set disableProcessingOfSuspendedTenants(value) {
+    this.#disableProcessingOfSuspendedTenants = value;
+  }
+
   /**
     @return { Config }
   **/

src/config.js

@@ -47,6 +47,7 @@ const CONFIG_VARS = [
   ["publishEventBlockList", true],
   ["crashOnRedisUnavailable", false],
   ["enableAdminService", false],
+  ["disableProcessingOfSuspendedTenants", true],
 ];
 
 /**

src/initialize.js

@@ -6,10 +6,13 @@ const cds = require("@sap/cds");
 const config = require("../config");
 const common = require("./common");
 const { TenantIdCheckTypes } = require("../constants");
+const { limiter } = require("./common");
 
 const VERROR_CLUSTER_NAME = "ExecuteInNewTransactionError";
 const COMPONENT_NAME = "/eventQueue/cdsHelper";
 
+const CONCURRENCY_AUTH_INFO = 3;
+
 /**
  * Execute logic in a new managed CDS transaction context, auto-handling commit, rollback and error/exception situations.
  * Includes logging of start, end and error situation with additional info object and unique transaction id (txId)
@@ -140,15 +143,26 @@ const getAllTenantIds = async () => {
     return null;
   }
 
-  return response
-    .map((tenant) => tenant.subscribedTenantId ?? tenant.tenant)
-    .reduce(async (result, tenantId) => {
-      result = await result;
-      if (await common.isTenantIdValidCb(TenantIdCheckTypes.eventProcessing, tenantId)) {
-        result.push(tenantId);
+  const tenantIds = response.map((tenant) => tenant.subscribedTenantId ?? tenant.tenant);
+  const suspendedTenants = {};
+  if (config.disableProcessingOfSuspendedTenants) {
+    await limiter(CONCURRENCY_AUTH_INFO, tenantIds, async (tenantId) => {
+      const result = await common.getAuthContext(tenantId, { returnError: true });
+      // NOTE: only 404 errors are propagated all others are ignored
+      if (result?.[0]) {
+        suspendedTenants[tenantId] = true;
+        cds.log(COMPONENT_NAME).info("skip event-queue processing, tenant suspended", { tenantId });
       }
-      return result;
-    }, []);
+    });
+  }
+
+  return tenantIds.reduce(async (result, tenantId) => {
+    result = await result;
+    if (!suspendedTenants[tenantId] && (await common.isTenantIdValidCb(TenantIdCheckTypes.eventProcessing, tenantId))) {
+      result.push(tenantId);
+    }
+    return result;
+  }, []);
 };
 
 const TENANT_COLUMNS = ["subscribedSubdomain", "createdAt", "modifiedAt"];

src/shared/cdsHelper.js

@@ -10,7 +10,8 @@ const config = require("../config");
 const { ExpiringLazyCache } = require("./lazyCache");
 const { TenantIdCheckTypes } = require("../constants");
 
-const MARGIN_AUTH_INFO_EXPIRY = 60 * 1000;
+const EXPIRE_TIME_TENANT_404 = 10 * 60 * 1000; // 10 minutes
+
 const COMPONENT_NAME = "/eventQueue/common";
 
 const arrayToFlatMap = (array, key = "ID") => {
@@ -97,18 +98,22 @@ const _getNewAuthContext = async (tenantId) => {
     const token = await authService.fetchClientCredentialsToken({ zid: tenantId });
     const tokenInfo = new xssec.XsuaaToken(token.access_token);
     const authInfo = new xssec.XsuaaSecurityContext(authService, tokenInfo);
-    return [tokenInfo.getExpirationDate().getTime() - Date.now(), authInfo];
+    return [tokenInfo.getExpirationDate().getTime() - Date.now(), [null, authInfo]];
   } catch (err) {
     cds.log(COMPONENT_NAME).warn("failed to request authContext", {
       err: err.message,
       responseCode: err.responseCode,
       responseText: err.responseText,
     });
+
+    if (err.responseCode === 404) {
+      return [EXPIRE_TIME_TENANT_404, [err, null]];
+    }
     return [0, null];
   }
 };
 
-const getAuthContext = async (tenantId) => {
+const getAuthContext = async (tenantId, { returnError = false } = {}) => {
   if (!(await isTenantIdValidCb(TenantIdCheckTypes.getAuthContext, tenantId))) {
     return null;
   }
@@ -125,8 +130,13 @@ const getAuthContext = async (tenantId) => {
     return null;
   }
 
-  getAuthContext._cache = getAuthContext._cache ?? new ExpiringLazyCache({ expirationGap: MARGIN_AUTH_INFO_EXPIRY });
-  return await getAuthContext._cache.getSetCb(tenantId, async () => _getNewAuthContext(tenantId));
+  getAuthContext._cache = getAuthContext._cache ?? new ExpiringLazyCache();
+  const result = await getAuthContext._cache.getSetCb(tenantId, async () => _getNewAuthContext(tenantId));
+  if (returnError) {
+    return result;
+  } else {
+    return result?.[1];
+  }
 };
 
 const isTenantIdValidCb = async (checkType, tenantId) => {

src/shared/common.js

@@ -1,7 +1,7 @@
 "use strict";
 
 const DEFAULT_SEPARATOR = "##";
-const DEFAULT_EXPIRATION_GAP = 5000; // 5 seconds
+const DEFAULT_EXPIRATION_GAP = 60 * 1000; // 60 seconds
 
 class LazyCache {
   constructor({ separator = DEFAULT_SEPARATOR } = {}) {

src/shared/lazyCache.js

@@ -140,14 +140,14 @@ describe("getAuthContext", () => {
   it("should handle error", async () => {
     jest.spyOn(xssec.XsuaaService.prototype, "fetchClientCredentialsToken").mockRejectedValueOnce(new Error());
     const result = await getAuthContext(tenantId1);
-    expect(result).toBeNull();
+    expect(result).toBeUndefined();
     expect(cds.log().warn.mock.calls).toMatchSnapshot();
   });
 
   it("should clear cache for error case", async () => {
     jest.spyOn(xssec.XsuaaService.prototype, "fetchClientCredentialsToken").mockRejectedValueOnce(new Error());
     const result = await getAuthContext(tenantId1);
-    expect(result).toBeNull();
+    expect(result).toBeUndefined();
     expect(cds.log().warn.mock.calls).toMatchSnapshot();
 
     jest