fix retry open after (#438)

Co-authored-by: Max Gruenfelder <maximilian.gruenfelder@scheer-group.com>

Author: soccermax

CHANGELOG.md

@@ -5,6 +5,17 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
+## v2.1.0 - 2025-02-XX
+
+### Added
+
+- [CAP Queue] Add support for defining successor and failed events of event handlers. See documentation for how to use it.
+- [CAP Queue] Allow to propagate cds.context properties (e.g. features). This can be configured per event (`cds.env.requires[<SERVICE>].queued.propagatedContextProperties = ["features"]`)
+
+### Fixed
+
+- The event property `retryOpenAfter` now works as expected.
+
 ## v2.0.4 - 2025-01-22
 
 ### Fixed

src/EventQueueProcessorBase.js

@@ -504,10 +504,16 @@ class EventQueueProcessorBase {
         if (!data.startAfter && [EventProcessingStatus.Error, EventProcessingStatus.Open].includes(data.status)) {
           data.startAfter = new Date(
             Date.now() +
-              (EventProcessingStatus.Error ? this.#eventConfig.retryFailedAfter : this.#eventConfig.retryOpenAfter)
+              (data.status === EventProcessingStatus.Error
+                ? this.#eventConfig.retryFailedAfter
+                : this.#eventConfig.retryOpenAfter)
           );
         }
 
+        if (data.status === EventProcessingStatus.Open && !("attempts" in data)) {
+          data.attempts = { "-=": 1 };
+        }
+
         if (data.startAfter) {
           this.#eventSchedulerInstance.scheduleEvent(
             this.__context.tenant,

test/asset/outboxProject/srv/service/standard-service.js

@@ -119,6 +119,14 @@ class StandardService extends cds.Service {
             }),
       }));
     });
+
+    this.on("retryOpenAfter", () => {
+      return { status: 0 };
+    });
+
+    this.on("retryFailedAfter", () => {
+      return { status: 3 };
+    });
   }
 }
 

test/eventQueueOutbox.test.js

@@ -126,6 +126,8 @@ cds.env.requires.StandardService = {
     propagateHeaders: ["authId", "customHeader"],
     events: {
       timeBucketAction: { timeBucket: "*/60 * * * * *" },
+      retryFailedAfter: { retryFailedAfter: 120 * 1000 },
+      retryOpenAfter: { retryOpenAfter: 140 * 1000 },
     },
   },
 };
@@ -2712,6 +2714,42 @@ describe("event-queue outbox", () => {
         });
       });
     });
+
+    describe("retry open and failed after", () => {
+      it("retry failed after", async () => {
+        const service = await cds.connect.to("StandardService");
+        await service.send("retryFailedAfter", {});
+        await commitAndOpenNew();
+        await processEventQueue(tx.context, "CAP_OUTBOX", `${service.name}.retryFailedAfter`);
+        const [event] = await testHelper.selectEventQueueAndReturn(tx, {
+          expectedLength: 1,
+          additionalColumns: ["lastAttemptTimestamp"],
+        });
+        expect(event).toMatchObject({
+          status: EventProcessingStatus.Error,
+          attempts: 1,
+          startAfter: new Date(new Date(event.lastAttemptTimestamp).getTime() + 120 * 1000).toISOString(),
+        });
+        expect(loggerMock.callsLengths().error).toEqual(0);
+      });
+
+      it("retry open after", async () => {
+        const service = await cds.connect.to("StandardService");
+        await service.send("retryOpenAfter", {});
+        await commitAndOpenNew();
+        await processEventQueue(tx.context, "CAP_OUTBOX", `${service.name}.retryOpenAfter`);
+        const [event] = await testHelper.selectEventQueueAndReturn(tx, {
+          expectedLength: 1,
+          additionalColumns: ["lastAttemptTimestamp"],
+        });
+        expect(event).toMatchObject({
+          status: EventProcessingStatus.Open,
+          attempts: 0,
+          startAfter: new Date(new Date(event.lastAttemptTimestamp).getTime() + 140 * 1000).toISOString(),
+        });
+        expect(loggerMock.callsLengths().error).toEqual(0);
+      });
+    });
   });
 
   const commitAndOpenNew = async () => {