Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: allow ignoring PublishError.Duplicate #404

Merged
merged 7 commits into from
Feb 21, 2023
Merged

feat: allow ignoring PublishError.Duplicate #404

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
f7f59cd
feat: allow for duplicate messages (#402)
maschad Feb 14, 2023
64780ed
feat: return empty array if ignoreDuplicateMessages is enabled (#402)
maschad Feb 15, 2023
6c8c0c2
feat: added gauge metric for tracking ignored duplicate messages (#402)
maschad Feb 15, 2023
2d30405
feat: updated metric name for gauge to include published messages tha…
maschad Feb 16, 2023
68c897b
chore: renaming
wemeetagain Feb 20, 2023
4b74381
chore: fix linter error
wemeetagain Feb 20, 2023
5906a27
chore: updated name to ignoreDuplicatePublishError (#402)
maschad Feb 20, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions src/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 */
allowPublishDuplicateMessages: 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 */
Expand Down Expand Up @@ -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 allowPublishDuplicateMessages = opts?.allowPublishDuplicateMessages ?? this.opts.allowPublishDuplicateMessages

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 (allowPublishDuplicateMessages) {
this.metrics?.onPublishDuplicateMsg(topic)
return { recipients: [] }
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@tuyennhv correct me if i'm wrong, this isn't what we want?
I thought we want to actually follow the same flow and actually send messages to peers?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

in this case some other nodes already published this same message and it's around in the topic peers already, I found it's not helpful to send to topic peers again - some nodes already saw that message and sent it to us, they may already had it in their seen cache too.

In lodestar's scenario, builder published the block, then the node receives the message and try to publish again and it throws error. The published block was spread to the network successfully, only our api throws error.

}
throw Error('PublishError.Duplicate')
}

Expand Down
11 changes: 11 additions & 0 deletions src/metrics.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -355,6 +355,12 @@ export function getMetrics(
labelNames: ['topic']
}),

duplicateMsgIgnored: register.gauge<{ topic: TopicLabel }>({
name: 'gossisub_published_duplicate_msgs_ignored_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({
Expand Down Expand Up @@ -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)
Expand Down
1 change: 1 addition & 0 deletions src/types.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,7 @@ export enum SignaturePolicy {

export type PublishOpts = {
allowPublishToZeroPeers?: boolean
allowPublishDuplicateMessages?: boolean
wemeetagain marked this conversation as resolved.
Show resolved Hide resolved
}

export enum PublishConfigType {
Expand Down