feat: Add retry logic for PubSub and Implement Deserialization generalized (#453)

* Add Deserializer

Signed-off-by: Xavier Geerinck <[email protected]>

* Convert Serializer to unified code base as well

Signed-off-by: Xavier Geerinck <[email protected]>

* Add Uint8Array deserialization

Signed-off-by: Xavier Geerinck <[email protected]>

* Add Uint8Array deserialization fix

Signed-off-by: Xavier Geerinck <[email protected]>

* Add implementation for PubSub retrial logic

Signed-off-by: Xavier Geerinck <[email protected]>

* Prettier

Signed-off-by: Xavier Geerinck <[email protected]>

* Cover linting issues

Signed-off-by: Xavier Geerinck <[email protected]>

* Changes

Signed-off-by: Xavier Geerinck <[email protected]>

* WIP

Signed-off-by: Xavier Geerinck <[email protected]>

* Add code with resilience

Signed-off-by: Xavier Geerinck <[email protected]>

* PRettier

Signed-off-by: Xavier Geerinck <[email protected]>

* Prettier, my best friend

Signed-off-by: Xavier Geerinck <[email protected]>

* Remove old comment block

Signed-off-by: Xavier Geerinck <[email protected]>

---------

Signed-off-by: Xavier Geerinck <[email protected]>

Author: XavierGeerinck

daprdocs/content/en/js-sdk-docs/js-server/_index.md

@@ -228,6 +228,77 @@ async function start() {
 
 > For a full list of state operations visit [How-To: Publish & subscribe]({{< ref howto-publish-subscribe.md >}}).
 
+#### Subscribe with SUCCESS/RETRY/DROP status
+
+Dapr supports [status codes for retry logic](https://docs.dapr.io/reference/api/pubsub_api/#expected-http-response) to specify what should happen after a message gets processed.
+
+> ⚠️ The JS SDK allows multiple callbacks on the same topic, we handle priority of status on `RETRY` > `DROP` > `SUCCESS` and default to `SUCCESS`
+
+> ⚠️ Make sure to [configure resiliency](https://docs.dapr.io/operations/resiliency/resiliency-overview/) in your application to handle `RETRY` messages
+
+In the JS SDK we support these messages through the `DaprPubSubStatusEnum` enum. To ensure Dapr will retry we configure a Resiliency policy as well.
+
+\*\*components/resiliency.yaml`
+
+```yaml
+apiVersion: dapr.io/v1alpha1
+kind: Resiliency
+metadata:
+  name: myresiliency
+spec:
+  policies:
+    retries:
+      # Global Retry Policy for Inbound Component operations
+      DefaultComponentInboundRetryPolicy:
+        policy: constant
+        duration: 500ms
+        maxRetries: 10
+  targets:
+    components:
+      messagebus:
+        inbound:
+          retry: DefaultComponentInboundRetryPolicy
+```
+
+**src/index.ts**
+
+```typescript
+import { DaprServer, DaprPubSubStatusEnum } from "@dapr/dapr";
+
+const daprHost = "127.0.0.1"; // Dapr Sidecar Host
+const daprPort = "3500"; // Dapr Sidecar Port of this Example Server
+const serverHost = "127.0.0.1"; // App Host of this Example Server
+const serverPort = "50051"; // App Port of this Example Server "
+
+async function start() {
+  const server = new DaprServer(serverHost, serverPort, daprHost, daprPort);
+
+  const pubSubName = "my-pubsub-name";
+  const topic = "topic-a";
+
+  // Process a message successfully
+  await server.pubsub.subscribe(pubSubName, topic, async (data: any, headers: object) => {
+    return DaprPubSubStatusEnum.SUCCESS;
+  });
+
+  // Retry a message
+  // Note: this example will keep on retrying to deliver the message
+  // Note 2: each component can have their own retry configuration
+  //   e.g., https://docs.dapr.io/reference/components-reference/supported-pubsub/setup-redis-pubsub/
+  await server.pubsub.subscribe(pubSubName, topic, async (data: any, headers: object) => {
+    return DaprPubSubStatusEnum.RETRY;
+  });
+
+  // Drop a message
+  await server.pubsub.subscribe(pubSubName, topic, async (data: any, headers: object) => {
+    return DaprPubSubStatusEnum.DROP;
+  });
+
+  // Start the server
+  await server.start();
+}
+```
+
 #### Subscribe to messages rule based
 
 Dapr [supports routing messages](https://docs.dapr.io/developing-applications/building-blocks/pubsub/howto-route-messages/) to different handlers (routes) based on rules.

src/enum/DaprPubSubStatus.enum.ts

@@ -0,0 +1,20 @@
+/*
+Copyright 2022 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+export enum DaprPubSubStatusEnum {
+  SUCCESS = "SUCCESS", // Message is processed successfully
+  RETRY = "RETRY", // Message to be retried by Dapr
+  DROP = "DROP", // Warning is logged and message is dropped
+}
+
+export default DaprPubSubStatusEnum;

src/implementation/Server/GRPCServer/GRPCServerImpl.ts

@@ -39,6 +39,9 @@ import { PubSubSubscriptionsType } from "../../../types/pubsub/PubSubSubscriptio
 import { DaprPubSubType } from "../../../types/pubsub/DaprPubSub.type";
 import { PubSubSubscriptionTopicRoutesType } from "../../../types/pubsub/PubSubSubscriptionTopicRoutes.type";
 import { DaprInvokerCallbackFunction } from "../../../types/DaprInvokerCallback.type";
+import { PubSubSubscriptionTopicRouteType } from "../../../types/pubsub/PubSubSubscriptionTopicRoute.type";
+import DaprPubSubStatusEnum from "../../../enum/DaprPubSubStatus.enum";
+import { deserializeGrpc } from "../../../utils/Deserializer.util";
 
 // https://github.com/badsyntax/grpc-js-typescript/issues/1#issuecomment-705419742
 // @ts-ignore
@@ -417,13 +420,7 @@ export default class GRPCServerImpl implements IAppCallbackServer {
       return;
     }
 
-    const data = Buffer.from(req.getData()).toString();
-    let dataParsed: any;
-    try {
-      dataParsed = JSON.parse(data);
-    } catch (_) {
-      dataParsed = data;
-    }
+    const data = deserializeGrpc(req.getDataContentType(), req.getData());
 
     const res = new TopicEventResponse();
 
@@ -436,19 +433,63 @@ export default class GRPCServerImpl implements IAppCallbackServer {
       }
     }
 
-    try {
-      const eventHandlers = this.pubSubSubscriptions[pubsubName][topic].routes[route].eventHandlers;
-      await Promise.all(eventHandlers.map((cb) => cb(dataParsed, headers)));
-      res.setStatus(TopicEventResponse.TopicEventResponseStatus.SUCCESS);
-    } catch (e) {
-      // @todo: for now we drop, maybe we should allow retrying as well more easily?
-      this.logger.error(`Error handling topic event: ${e}`);
-      res.setStatus(TopicEventResponse.TopicEventResponseStatus.DROP);
+    // Process the callbacks
+    // we handle priority of status on `RETRY` > `DROP` > `SUCCESS` and default to `SUCCESS`
+    const routeObj = this.pubSubSubscriptions[pubsubName][topic].routes[route];
+    const status = await this.processPubSubCallbacks(routeObj, data, headers);
+
+    switch (status) {
+      case DaprPubSubStatusEnum.RETRY:
+        res.setStatus(TopicEventResponse.TopicEventResponseStatus.RETRY);
+        break;
+      case DaprPubSubStatusEnum.DROP:
+        res.setStatus(TopicEventResponse.TopicEventResponseStatus.DROP);
+        break;
+      case DaprPubSubStatusEnum.SUCCESS:
+      default:
+        res.setStatus(TopicEventResponse.TopicEventResponseStatus.SUCCESS);
+        break;
     }
 
     return callback(null, res);
   }
 
+  async processPubSubCallbacks(
+    routeObj: PubSubSubscriptionTopicRouteType,
+    data: any,
+    headers: { [key: string]: string },
+  ): Promise<DaprPubSubStatusEnum> {
+    const eventHandlers = routeObj.eventHandlers;
+    const statuses = [];
+
+    // Process the callbacks (default: SUCCESS)
+    for (const cb of eventHandlers) {
+      let status = DaprPubSubStatusEnum.SUCCESS;
+
+      try {
+        status = await cb(data, headers);
+      } catch (e) {
+        // We catch and log an error, but we don't do anything with it as the statuses should define that
+        this.logger.error(`[route-${routeObj.path}] Message processing failed, ${e}`);
+      }
+
+      statuses.push(status ?? DaprPubSubStatusEnum.SUCCESS);
+    }
+
+    // Look at the statuses and return the highest priority
+    // we handle priority of status on `RETRY` > `DROP` > `SUCCESS`
+    if (statuses.includes(DaprPubSubStatusEnum.RETRY)) {
+      this.logger.debug(`[route-${routeObj.path}] Retrying message`);
+      return DaprPubSubStatusEnum.RETRY;
+    } else if (statuses.includes(DaprPubSubStatusEnum.DROP)) {
+      this.logger.debug(`[route-${routeObj.path}] Dropping message`);
+      return DaprPubSubStatusEnum.DROP;
+    } else {
+      this.logger.debug(`[route-${routeObj.path}] Acknowledging message`);
+      return DaprPubSubStatusEnum.SUCCESS;
+    }
+  }
+
   // Dapr will call this on startup to see which topics it is subscribed to
   async listTopicSubscriptions(
     call: grpc.ServerUnaryCall<Empty, ListTopicSubscriptionsResponse>,

src/implementation/Server/HTTPServer/HTTPServerImpl.ts

@@ -11,7 +11,6 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-import SubscribedMessageHttpResponse from "../../../enum/SubscribedMessageHttpResponse.enum";
 import { Logger } from "../../../logger/Logger";
 import { LoggerOptions } from "../../../types/logger/LoggerOptions";
 import { DaprPubSubType } from "../../../types/pubsub/DaprPubSub.type";
@@ -20,6 +19,9 @@ import { PubSubSubscriptionOptionsType } from "../../../types/pubsub/PubSubSubsc
 import { PubSubSubscriptionTopicRoutesType } from "../../../types/pubsub/PubSubSubscriptionTopicRoutes.type";
 import { IServerType } from "./HTTPServer";
 import { TypeDaprPubSubCallback } from "../../../types/DaprPubSubCallback.type";
+import DaprPubSubStatusEnum from "../../../enum/DaprPubSubStatus.enum";
+import { IncomingHttpHeaders } from "http";
+import { PubSubSubscriptionTopicRouteType } from "../../../types/pubsub/PubSubSubscriptionTopicRoute.type";
 
 export default class HTTPServerImpl {
   private readonly PUBSUB_DEFAULT_ROUTE_NAME = "default";
@@ -121,21 +123,12 @@ export default class HTTPServerImpl {
       // Add a server POST handler
       this.server.post(`/${routeObj.path}`, async (req, res) => {
         const headers = req.headers;
-
         const data = this.extractDataFromSubscribeRequest(req);
 
-        // Process the callback
-        try {
-          const eventHandlers = routeObj.eventHandlers;
-          await Promise.all(eventHandlers.map((cb) => cb(data, headers)));
-        } catch (e) {
-          this.logger.error(`[route-${routeObj.path}] Message processing failed, dropping: ${e}`);
-          return res.send({ status: SubscribedMessageHttpResponse.DROP });
-        }
-
-        // Let Dapr know that the message was processed correctly
-        this.logger.debug(`[route-${routeObj.path}] Acknowledging message`);
-        return res.send({ status: SubscribedMessageHttpResponse.SUCCESS });
+        // Process the callbacks
+        // we handle priority of status on `RETRY` > `DROP` > `SUCCESS` and default to `SUCCESS`
+        const status = await this.processPubSubCallbacks(routeObj, data, headers);
+        return res.send({ status });
       });
     }
 
@@ -146,6 +139,42 @@ export default class HTTPServerImpl {
     );
   }
 
+  async processPubSubCallbacks(
+    routeObj: PubSubSubscriptionTopicRouteType,
+    data: any,
+    headers: IncomingHttpHeaders,
+  ): Promise<DaprPubSubStatusEnum> {
+    const eventHandlers = routeObj.eventHandlers;
+    const statuses = [];
+
+    // Process the callbacks (default: SUCCESS)
+    for (const cb of eventHandlers) {
+      let status = DaprPubSubStatusEnum.SUCCESS;
+
+      try {
+        status = await cb(data, headers);
+      } catch (e) {
+        // We catch and log an error, but we don't do anything with it as the statuses should define that
+        this.logger.error(`[route-${routeObj.path}] Message processing failed, ${e}`);
+      }
+
+      statuses.push(status ?? DaprPubSubStatusEnum.SUCCESS);
+    }
+
+    // Look at the statuses and return the highest priority
+    // we handle priority of status on `RETRY` > `DROP` > `SUCCESS`
+    if (statuses.includes(DaprPubSubStatusEnum.RETRY)) {
+      this.logger.debug(`[route-${routeObj.path}] Retrying message`);
+      return DaprPubSubStatusEnum.RETRY;
+    } else if (statuses.includes(DaprPubSubStatusEnum.DROP)) {
+      this.logger.debug(`[route-${routeObj.path}] Dropping message`);
+      return DaprPubSubStatusEnum.DROP;
+    } else {
+      this.logger.debug(`[route-${routeObj.path}] Acknowledging message`);
+      return DaprPubSubStatusEnum.SUCCESS;
+    }
+  }
+
   registerPubSubSubscriptionEventHandler(
     pubsubName: string,
     topic: string,

src/index.ts

@@ -11,25 +11,32 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-import HttpMethod from "./enum/HttpMethod.enum";
-import CommunicationProtocolEnum from "./enum/CommunicationProtocol.enum";
-import DaprClient from "./implementation/Client/DaprClient";
-import DaprServer from "./implementation/Server/DaprServer";
-import AbstractActor from "./actors/runtime/AbstractActor";
 import { Temporal } from "@js-temporal/polyfill";
-import ActorProxyBuilder from "./actors/client/ActorProxyBuilder";
+
+import DaprServer from "./implementation/Server/DaprServer";
+import DaprClient from "./implementation/Client/DaprClient";
+import GRPCClient from "./implementation/Client/GRPCClient/GRPCClient";
+import HTTPClient from "./implementation/Client/HTTPClient/HTTPClient";
+
 import ActorId from "./actors/ActorId";
+import ActorProxyBuilder from "./actors/client/ActorProxyBuilder";
+import AbstractActor from "./actors/runtime/AbstractActor";
+
+import { ConsoleLoggerService } from "./logger/ConsoleLoggerService";
+
 import { DaprClientOptions } from "./types/DaprClientOptions";
+import { InvokerOptions } from "./types/InvokerOptions.type";
+import { DaprInvokerCallbackContent, DaprInvokerCallbackFunction } from "./types/DaprInvokerCallback.type";
 import { LoggerOptions } from "./types/logger/LoggerOptions";
 import { LoggerService } from "./types/logger/LoggerService";
-import { ConsoleLoggerService } from "./logger/ConsoleLoggerService";
 import { LogLevel } from "./types/logger/LogLevel";
-import GRPCClient from "./implementation/Client/GRPCClient/GRPCClient";
-import HTTPClient from "./implementation/Client/HTTPClient/HTTPClient";
-import { InvokerOptions } from "./types/InvokerOptions.type";
-import { DaprInvokerCallbackContent, DaprInvokerCallbackFunction } from "./types/DaprInvokerCallback.type";
 import { PubSubBulkPublishResponse } from "./types/pubsub/PubSubBulkPublishResponse.type";
 import { PubSubBulkPublishMessage } from "./types/pubsub/PubSubBulkPublishMessage.type";
+
+import HttpMethod from "./enum/HttpMethod.enum";
+import CommunicationProtocolEnum from "./enum/CommunicationProtocol.enum";
+import DaprPubSubStatusEnum from "./enum/DaprPubSubStatus.enum";
+
 export {
   DaprClient,
   DaprServer,
@@ -39,7 +46,6 @@ export {
   AbstractActor,
   ActorId,
   ActorProxyBuilder,
-  CommunicationProtocolEnum,
   Temporal,
   DaprClientOptions,
   LogLevel,
@@ -49,6 +55,8 @@ export {
   InvokerOptions,
   DaprInvokerCallbackFunction as TypeDaprInvokerCallback,
   DaprInvokerCallbackContent,
+  CommunicationProtocolEnum,
+  DaprPubSubStatusEnum,
   PubSubBulkPublishMessage,
   PubSubBulkPublishResponse,
 };

src/utils/Client.util.ts

@@ -11,13 +11,16 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
+import { randomUUID } from "crypto";
+import { Map } from "google-protobuf";
+
+import { ConfigurationItem as ConfigurationItemProto } from "../proto/dapr/proto/common/v1/common_pb";
+import { isCloudEvent } from "./CloudEvent.util";
+
 import { KeyValueType } from "../types/KeyValue.type";
 import { ConfigurationType } from "../types/configuration/Configuration.type";
 import { ConfigurationItem } from "../types/configuration/ConfigurationItem";
-import { ConfigurationItem as ConfigurationItemProto } from "../proto/dapr/proto/common/v1/common_pb";
-import { Map } from "google-protobuf";
 import { PubSubBulkPublishEntry } from "../types/pubsub/PubSubBulkPublishEntry.type";
-import { randomUUID } from "crypto";
 import { PubSubBulkPublishResponse } from "../types/pubsub/PubSubBulkPublishResponse.type";
 import { PubSubBulkPublishMessage } from "../types/pubsub/PubSubBulkPublishMessage.type";
 import { PubSubBulkPublishApiResponse } from "../types/pubsub/PubSubBulkPublishApiResponse.type";
@@ -82,24 +85,6 @@ export function createConfigurationType(configDict: Map<string, ConfigurationIte
   return configMap;
 }
 
-/**
- * Checks if the input object is a valid Cloud Event.
- * A valid Cloud Event is a JSON object that contains id, source, type, and specversion.
- * See https://github.com/cloudevents/spec/blob/v1.0/spec.md#required-attributes
- * @param str input object
- * @returns true if the object is a valid Cloud Event
- */
-function isCloudEvent(obj: object): boolean {
-  const requiredAttributes = ["id", "source", "type", "specversion"];
-  return (
-    typeof obj === "object" &&
-    obj !== null &&
-    requiredAttributes.every((attr) => {
-      return Object.prototype.hasOwnProperty.call(obj, attr);
-    })
-  );
-}
-
 /**
  * Gets the Content-Type for the input data.
  *