Fb/discover cds events (#412)

* discover cds events from env

* discover cds events from env

* fixes and test improvements

* fixes and test improvements

* fixes and test improvements

* wip

* add tests and fixes for delayed events

* add tests and fixes for delayed events

* wip

* wip

* wip

* wip

* wip

* wip

* wip

* wip

* fix

Author: soccermax

CHANGELOG.md

@@ -7,10 +7,15 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## v2.0.2 - 2025-11-XX
 
+### Added
+
+- [CAP Queue] Configuration is initialized by CDS bootstrap instead of lazy initialization on first use.
+
 ### Fixed
 
 - [Admin Service] Correctly return open locks after adding namespace to locks
 - [Admin Service] Publish events improved error handling
+- [Event Processing] In some cases where more delayed events than the `selectMaxChunkSize` existed, the processing of open events was delayed.
 
 ## v2.0.1 - 2025-11-24
 

package-lock.json

@@ -56,10 +56,11 @@
   "devDependencies": {
     "@actions/core": "^1.11.1",
     "@cap-js/cds-test": "^0.4.0",
+    "@cap-js/db-service": "^2.6.0",
     "@cap-js/hana": "^2.3.4",
-    "@cap-js/sqlite": "^2.0.4",
+    "@cap-js/sqlite": "^2.1.0",
     "@opentelemetry/api": "^1.9.0",
-    "@sap/cds": "^9.4.4",
+    "@sap/cds": "^9.4.5",
     "@sap/cds-dk": "^9.4.2",
     "eslint": "^8.57.0",
     "eslint-config-prettier": "^9.1.0",

package.json

@@ -27,12 +27,16 @@ const ERROR_CODES = {
   APP_NAMES_FORMAT: "APP_NAMES_FORMAT",
   APP_INSTANCES_FORMAT: "APP_INSTANCES_FORMAT",
   MULTI_INSTANCE_PROCESSING_NOT_ALLOWED: "MULTI_INSTANCE_PROCESSING_NOT_ALLOWED",
+  INVALID_CLUSTER_HANDLER_RESULT: "INVALID_CLUSTER_HANDLER_RESULT",
 };
 
 const ERROR_CODES_META = {
   [ERROR_CODES.WRONG_TX_USAGE]: {
     message: "Usage of this.tx|this.context is not allowed if parallel event processing is enabled",
   },
+  [ERROR_CODES.INVALID_CLUSTER_HANDLER_RESULT]: {
+    message: "A cluster handler returned an invalid result. Cluster handlers must return the clustered payload data.",
+  },
   [ERROR_CODES.UNKNOWN_EVENT_TYPE]: {
     message: "The event type and subType configuration is not configured! Maintain the combination in the config file.",
   },
@@ -371,6 +375,17 @@ class EventQueueError extends VError {
     );
   }
 
+  static invalidClusterHandlerResult(clusterKey, propertyName) {
+    const { message } = ERROR_CODES_META[ERROR_CODES.INVALID_CLUSTER_HANDLER_RESULT];
+    return new EventQueueError(
+      {
+        name: ERROR_CODES.INVALID_CLUSTER_HANDLER_RESULT,
+        info: { clusterKey, propertyName },
+      },
+      message
+    );
+  }
+
   static isRedisConnectionFailure(err) {
     return err instanceof VError && err.name === ERROR_CODES.REDIS_CREATE_CLIENT;
   }

src/EventQueueError.js

@@ -62,6 +62,7 @@ class EventQueueProcessorBase {
     this.#eventType = eventType;
     this.#eventSubType = eventSubType;
     this.#eventConfig = config ?? {};
+    this.#eventConfig.selectedDelayedEventIds ??= [];
     this.__parallelEventProcessing = this.#eventConfig.parallelEventProcessing ?? DEFAULT_PARALLEL_EVENT_PROCESSING;
     if (this.__parallelEventProcessing > LIMIT_PARALLEL_EVENT_PROCESSING) {
       this.__parallelEventProcessing = LIMIT_PARALLEL_EVENT_PROCESSING;
@@ -664,45 +665,49 @@ class EventQueueProcessorBase {
     const baseDate = Date.now();
     const refDateStartAfter = new Date(baseDate + this.#config.runInterval * 1.2);
     await executeInNewTransaction(this.__baseContext, "eventQueue-getQueueEntriesAndSetToInProgress", async (tx) => {
-      const entries = await tx.run(
-        SELECT.from(this.#config.tableNameEventQueue)
-          .forUpdate({ wait: this.#config.forUpdateTimeout })
-          .limit(this.selectMaxChunkSize)
-          .where(
-            "type =",
-            this.#eventType,
-            "AND subType=",
-            this.#eventSubType,
-            "AND namespace =",
-            this.#namespace,
-            "AND ( startAfter IS NULL OR startAfter <=",
-            refDateStartAfter.toISOString(),
-            " ) AND ( status =",
-            EventProcessingStatus.Open,
-            "AND ( lastAttemptTimestamp <=",
-            this.startTime.toISOString(),
-            ...(this.isPeriodicEvent
-              ? [
-                  "OR lastAttemptTimestamp IS NULL ) OR ( status =",
-                  EventProcessingStatus.InProgress,
-                  "AND lastAttemptTimestamp <=",
-                  new Date(baseDate - this.#eventConfig.keepAliveMaxInProgressTime * 1000).toISOString(),
-                  ") )",
-                ]
-              : [
-                  "OR lastAttemptTimestamp IS NULL ) OR ( status =",
-                  EventProcessingStatus.Error,
-                  "AND lastAttemptTimestamp <=",
-                  this.startTime.toISOString(),
-                  ") OR ( status =",
-                  EventProcessingStatus.InProgress,
-                  "AND lastAttemptTimestamp <=",
-                  new Date(baseDate - this.#eventConfig.keepAliveMaxInProgressTime * 1000).toISOString(),
-                  ") )",
-                ])
-          )
-          .orderBy("createdAt", "ID")
-      );
+      const cqn = SELECT.from(this.#config.tableNameEventQueue)
+        .forUpdate({ wait: this.#config.forUpdateTimeout })
+        .limit(this.selectMaxChunkSize)
+        .where(
+          "type =",
+          this.#eventType,
+          "AND subType=",
+          this.#eventSubType,
+          "AND namespace =",
+          this.#namespace,
+          "AND ( startAfter IS NULL OR startAfter <=",
+          refDateStartAfter.toISOString(),
+          " ) AND ( status =",
+          EventProcessingStatus.Open,
+          "AND ( lastAttemptTimestamp <=",
+          this.startTime.toISOString(),
+          ...(this.isPeriodicEvent
+            ? [
+                "OR lastAttemptTimestamp IS NULL ) OR ( status =",
+                EventProcessingStatus.InProgress,
+                "AND lastAttemptTimestamp <=",
+                new Date(baseDate - this.#eventConfig.keepAliveMaxInProgressTime * 1000).toISOString(),
+                ") ) ",
+              ]
+            : [
+                "OR lastAttemptTimestamp IS NULL ) OR ( status =",
+                EventProcessingStatus.Error,
+                "AND lastAttemptTimestamp <=",
+                this.startTime.toISOString(),
+                ") OR ( status =",
+                EventProcessingStatus.InProgress,
+                "AND lastAttemptTimestamp <=",
+                new Date(baseDate - this.#eventConfig.keepAliveMaxInProgressTime * 1000).toISOString(),
+                ") )",
+              ])
+        )
+        .orderBy("createdAt", "ID");
+
+      if (this.#eventConfig.selectedDelayedEventIds) {
+        cqn.where("ID NOT IN", this.#eventConfig.selectedDelayedEventIds);
+      }
+
+      const entries = await tx.run(cqn);
 
       if (!entries.length) {
         this.logger.debug("no entries available for processing", {
@@ -756,7 +761,9 @@ class EventQueueProcessorBase {
       }
 
       if (!eventsForProcessing.length) {
-        this.__emptyChunkSelected = true;
+        if (!entries.length) {
+          this.__emptyChunkSelected = true;
+        }
         return;
       }
 
@@ -806,8 +813,11 @@ class EventQueueProcessorBase {
     return entry.lastAttemptsTs;
   }
 
-  #handleDelayedEvents(delayedEvents) {
+  #handleDelayedEvents(delayedEvents, { skipExcludeDelayedEventIds = false } = {}) {
     for (const delayedEvent of delayedEvents) {
+      if (!skipExcludeDelayedEventIds) {
+        this.#eventConfig.selectedDelayedEventIds.push(delayedEvent.ID);
+      }
       this.#eventSchedulerInstance.scheduleEvent(
         this.__context.tenant,
         this.#eventType,
@@ -1107,6 +1117,7 @@ class EventQueueProcessorBase {
     }
 
     const newEvent = {
+      ID: cds.utils.uuid(),
       type: this.#eventType,
       subType: this.#eventSubType,
       namespace: this.#eventConfig.namespace,
@@ -1131,16 +1142,16 @@ class EventQueueProcessorBase {
       });
     }
 
-    this.tx._skipEventQueueBroadcase = true;
+    this.tx._skipEventQueueBroadcast = true;
     await this.tx.run(
       INSERT.into(this.#config.tableNameEventQueue).entries({
         ...newEvent,
         startAfter: newEvent.startAfter.toISOString(),
       })
     );
-    this.tx._skipEventQueueBroadcase = false;
+    this.tx._skipEventQueueBroadcast = false;
     if (intervalInMs < this.#config.runInterval * 1.5) {
-      this.#handleDelayedEvents([newEvent]);
+      this.#handleDelayedEvents([newEvent], { skipExcludeDelayedEventIds: true });
       const { relative: relativeAfterSchedule } = this.#eventSchedulerInstance.calculateOffset(
         this.#eventType,
         this.#eventSubType,