From dcde3c97ee9eee8c8f1c9c496a5232ea43b21d37 Mon Sep 17 00:00:00 2001
From: Chad Nehemiah <chad.nehemiah94@gmail.com>
Date: Mon, 20 Feb 2023 20:00:14 -0500
Subject: [PATCH] feat: allow ignoring PublishError.Duplicate (#404)

* feat: allow for duplicate messages (#402)

* feat: return empty array if ignoreDuplicateMessages is enabled (#402)

* feat: added gauge metric for tracking ignored duplicate messages (#402)

* feat: updated metric name for gauge to include published messages that are ignored (#402)

* chore: renaming

* chore: fix linter error

* chore: updated name to ignoreDuplicatePublishError (#402)

---------

Co-authored-by: Cayman <caymannava@gmail.com>
---
 src/index.ts   |  9 +++++++++
 src/metrics.ts | 11 +++++++++++
 src/types.ts   |  1 +
 3 files changed, 21 insertions(+)

diff --git a/src/index.ts b/src/index.ts
index 2b8c044c..27a632f7 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -109,6 +109,8 @@ export interface GossipsubOpts extends GossipsubOptsSpec, PubSubInit {
   asyncValidation: boolean
   /** Do not throw `InsufficientPeers` error if publishing to zero peers */
   allowPublishToZeroPeers: boolean
+  /** Do not throw `PublishError.Duplicate` if publishing duplicate messages */
+  ignoreDuplicatePublishError: boolean
   /** For a single stream, await processing each RPC before processing the next */
   awaitRpcHandler: boolean
   /** For a single RPC, await processing each message before processing the next */
@@ -2013,9 +2015,16 @@ export class GossipSub extends EventEmitter<GossipsubEvents> implements PubSub<G
     const msgId = await this.msgIdFn(msg)
     const msgIdStr = this.msgIdToStrFn(msgId)
 
+    // Current publish opt takes precedence global opts, while preserving false value
+    const ignoreDuplicatePublishError = opts?.ignoreDuplicatePublishError ?? this.opts.ignoreDuplicatePublishError
+
     if (this.seenCache.has(msgIdStr)) {
       // This message has already been seen. We don't re-publish messages that have already
       // been published on the network.
+      if (ignoreDuplicatePublishError) {
+        this.metrics?.onPublishDuplicateMsg(topic)
+        return { recipients: [] }
+      }
       throw Error('PublishError.Duplicate')
     }
 
diff --git a/src/metrics.ts b/src/metrics.ts
index caa29ba6..d7fd2824 100644
--- a/src/metrics.ts
+++ b/src/metrics.ts
@@ -355,6 +355,12 @@ export function getMetrics(
       labelNames: ['topic']
     }),
 
+    duplicateMsgIgnored: register.gauge<{ topic: TopicLabel }>({
+      name: 'gossisub_ignored_published_duplicate_msgs_total',
+      help: 'Total count of published duplicate message ignored by topic',
+      labelNames: ['topic']
+    }),
+
     /* Metrics related to scoring */
     /** Total times score() is called */
     scoreFnCalls: register.gauge({
@@ -629,6 +635,11 @@ export function getMetrics(
       }
     },
 
+    onPublishDuplicateMsg(topicStr: TopicStr): void {
+      const topic = this.toTopic(topicStr)
+      this.duplicateMsgIgnored.inc({ topic }, 1)
+    },
+
     onRpcRecv(rpc: IRPC, rpcBytes: number): void {
       this.rpcRecvBytes.inc(rpcBytes)
       this.rpcRecvCount.inc(1)
diff --git a/src/types.ts b/src/types.ts
index 714b5419..32c2fd29 100644
--- a/src/types.ts
+++ b/src/types.ts
@@ -74,6 +74,7 @@ export enum SignaturePolicy {
 
 export type PublishOpts = {
   allowPublishToZeroPeers?: boolean
+  ignoreDuplicatePublishError?: boolean
 }
 
 export enum PublishConfigType {