Skip to content

Commit 9652f24

Browse files
smrz2001lidel
smrz2001
and
lidel
authored
feat: Pubsub.SeenMessagesStrategy (#9543)
* feat: expire messages from the cache based on last seen time * docs: Pubsub.SeenMessagesStrategy Ref. libp2p/go-libp2p-pubsub#513 Co-authored-by: Marcin Rataj <[email protected]>
1 parent f20c980 commit 9652f24

File tree

9 files changed

+191
-53
lines changed

9 files changed

+191
-53
lines changed

‎config/pubsub.go

Lines changed: 25 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,24 @@
11
package config
22

3+
const (
4+
// LastSeenMessagesStrategy is a strategy that calculates the TTL countdown
5+
// based on the last time a Pubsub message is seen. This means that if a message
6+
// is received and then seen again within the specified TTL window, it
7+
// won't be emitted until the TTL countdown expires from the last time the
8+
// message was seen.
9+
LastSeenMessagesStrategy = "last-seen"
10+
11+
// FirstSeenMessagesStrategy is a strategy that calculates the TTL
12+
// countdown based on the first time a Pubsub message is seen. This means that if
13+
// a message is received and then seen again within the specified TTL
14+
// window, it won't be emitted.
15+
FirstSeenMessagesStrategy = "first-seen"
16+
17+
// DefaultSeenMessagesStrategy is the strategy that is used by default if
18+
// no Pubsub.SeenMessagesStrategy is specified.
19+
DefaultSeenMessagesStrategy = LastSeenMessagesStrategy
20+
)
21+
322
type PubsubConfig struct {
423
// Router can be either floodsub (legacy) or gossipsub (new and
524
// backwards compatible).
@@ -12,7 +31,11 @@ type PubsubConfig struct {
1231
// Enable pubsub (--enable-pubsub-experiment)
1332
Enabled Flag `json:",omitempty"`
1433

15-
// SeenMessagesTTL configures the duration after which a previously seen
16-
// message ID can be forgotten about.
34+
// SeenMessagesTTL is a value that controls the time window within which
35+
// duplicate messages will be identified and won't be emitted.
1736
SeenMessagesTTL *OptionalDuration `json:",omitempty"`
37+
38+
// SeenMessagesStrategy is a setting that determines how the time-to-live
39+
// (TTL) countdown for deduplicating messages is calculated.
40+
SeenMessagesStrategy *OptionalString `json:",omitempty"`
1841
}

‎core/node/groups.go

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import (
1111
"github.com/ipfs/go-log"
1212
"github.com/ipfs/kubo/config"
1313
pubsub "github.com/libp2p/go-libp2p-pubsub"
14+
"github.com/libp2p/go-libp2p-pubsub/timecache"
1415
"github.com/libp2p/go-libp2p/core/peer"
1516

1617
"github.com/ipfs/kubo/core/node/libp2p"
@@ -66,6 +67,18 @@ func LibP2P(bcfg *BuildCfg, cfg *config.Config) fx.Option {
6667
pubsub.WithSeenMessagesTTL(cfg.Pubsub.SeenMessagesTTL.WithDefault(pubsub.TimeCacheDuration)),
6768
)
6869

70+
var seenMessagesStrategy timecache.Strategy
71+
configSeenMessagesStrategy := cfg.Pubsub.SeenMessagesStrategy.WithDefault(config.DefaultSeenMessagesStrategy)
72+
switch configSeenMessagesStrategy {
73+
case config.LastSeenMessagesStrategy:
74+
seenMessagesStrategy = timecache.Strategy_LastSeen
75+
case config.FirstSeenMessagesStrategy:
76+
seenMessagesStrategy = timecache.Strategy_FirstSeen
77+
default:
78+
return fx.Error(fmt.Errorf("unsupported Pubsub.SeenMessagesStrategy %q", configSeenMessagesStrategy))
79+
}
80+
pubsubOptions = append(pubsubOptions, pubsub.WithSeenMessagesStrategy(seenMessagesStrategy))
81+
6982
switch cfg.Pubsub.Router {
7083
case "":
7184
fallthrough

‎docs/changelogs/v0.18.md

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,43 @@
11
# Kubo changelog v0.18
22

3+
## v0.18.1
4+
5+
This release includes improvements around Pubsub message deduplication, and more.
6+
7+
8+
<!-- TOC depthfrom:3 -->
9+
10+
- [Overview](#overview)
11+
- [🔦 Highlights](#-highlights)
12+
- [New default Pubsub.SeenMessagesStrategy](#new-default-pubsubseenmessagesstrategy)
13+
- [📝 Changelog](#-changelog)
14+
- [👨‍👩‍👧‍👦 Contributors](#-contributors)
15+
16+
<!-- /TOC -->
17+
18+
### 🔦 Highlights
19+
20+
#### New default `Pubsub.SeenMessagesStrategy`
21+
22+
A new optional [`Pubsub.SeenMessagesStrategy`](../config.md#pubsubseenmessagesstrategy) configuration option has been added.
23+
24+
This option allows you to choose between two different strategies for
25+
deduplicating messages: `first-seen` and `last-seen`.
26+
27+
When unset, the default strategy is `last-seen`, which calculates the
28+
time-to-live (TTL) countdown based on the last time a message is seen. This
29+
means that if a message is received and then seen again within the specified
30+
TTL window based on the last time it was seen, it won't be emitted.
31+
32+
If you prefer the old behavior, which calculates the TTL countdown based on the
33+
first time a message is seen, you can set `Pubsub.SeenMessagesStrategy` to
34+
`first-seen`.
35+
36+
### 📝 Changelog
37+
38+
### 👨‍👩‍👧‍👦 Contributors
39+
40+
341
## v0.18.0
442

543
### Overview

‎docs/config.md

Lines changed: 26 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -100,6 +100,7 @@ config file at runtime.
100100
- [`Pubsub.Router`](#pubsubrouter)
101101
- [`Pubsub.DisableSigning`](#pubsubdisablesigning)
102102
- [`Pubsub.SeenMessagesTTL`](#pubsubseenmessagesttl)
103+
- [`Pubsub.SeenMessagesStrategy`](#pubsubseenmessagesstrategy)
103104
- [`Peering`](#peering)
104105
- [`Peering.Peers`](#peeringpeers)
105106
- [`Reprovider`](#reprovider)
@@ -1206,8 +1207,8 @@ Type: `bool`
12061207

12071208
### `Pubsub.SeenMessagesTTL`
12081209

1209-
Configures the duration after which a previously seen Pubsub Message ID can be
1210-
forgotten about.
1210+
Controls the time window within which duplicate messages, identified by Message
1211+
ID, will be identified and won't be emitted again.
12111212

12121213
A smaller value for this parameter means that Pubsub messages in the cache will
12131214
be garbage collected sooner, which can result in a smaller cache. At the same
@@ -1223,6 +1224,29 @@ Default: see `TimeCacheDuration` from [go-libp2p-pubsub](https://github.com/libp
12231224

12241225
Type: `optionalDuration`
12251226

1227+
### `Pubsub.SeenMessagesStrategy`
1228+
1229+
Determines how the time-to-live (TTL) countdown for deduplicating Pubsub
1230+
messages is calculated.
1231+
1232+
The Pubsub seen messages cache is a LRU cache that keeps messages for up to a
1233+
specified time duration. After this duration has elapsed, expired messages will
1234+
be purged from the cache.
1235+
1236+
The `last-seen` cache is a sliding-window cache. Every time a message is seen
1237+
again with the SeenMessagesTTL duration, its timestamp slides forward. This
1238+
keeps frequently occurring messages cached and prevents them from being
1239+
continually propagated, especially because of issues that might increase the
1240+
number of duplicate messages in the network.
1241+
1242+
The `first-seen` cache will store new messages and purge them after the
1243+
SeenMessagesTTL duration, even if they are seen multiple times within this
1244+
duration.
1245+
1246+
Default: `last-seen` (see [go-libp2p-pubsub](https://github.com/libp2p/go-libp2p-pubsub))
1247+
1248+
Type: `optionalString`
1249+
12261250
## `Peering`
12271251

12281252
Configures the peering subsystem. The peering subsystem configures Kubo to

‎docs/examples/kubo-as-a-library/go.mod

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@ require (
3838
github.com/docker/go-units v0.5.0 // indirect
3939
github.com/dustin/go-humanize v1.0.0 // indirect
4040
github.com/elastic/gosigar v0.14.2 // indirect
41+
github.com/emirpasic/gods v1.18.1 // indirect
4142
github.com/facebookgo/atomicfile v0.0.0-20151019160806-2de1f203e7d5 // indirect
4243
github.com/flynn/noise v1.0.0 // indirect
4344
github.com/francoispqt/gojay v1.2.13 // indirect
@@ -122,7 +123,7 @@ require (
122123
github.com/libp2p/go-libp2p-asn-util v0.2.0 // indirect
123124
github.com/libp2p/go-libp2p-kad-dht v0.20.0 // indirect
124125
github.com/libp2p/go-libp2p-kbucket v0.5.0 // indirect
125-
github.com/libp2p/go-libp2p-pubsub v0.8.2 // indirect
126+
github.com/libp2p/go-libp2p-pubsub v0.8.3 // indirect
126127
github.com/libp2p/go-libp2p-pubsub-router v0.6.0 // indirect
127128
github.com/libp2p/go-libp2p-record v0.2.0 // indirect
128129
github.com/libp2p/go-libp2p-routing-helpers v0.6.0 // indirect
@@ -180,7 +181,6 @@ require (
180181
github.com/whyrusleeping/chunker v0.0.0-20181014151217-fe64bd25879f // indirect
181182
github.com/whyrusleeping/go-keyspace v0.0.0-20160322163242-5b898ac5add1 // indirect
182183
github.com/whyrusleeping/multiaddr-filter v0.0.0-20160516205228-e903e4adabd7 // indirect
183-
github.com/whyrusleeping/timecache v0.0.0-20160911033111-cfcb2f1abfee // indirect
184184
go.opencensus.io v0.24.0 // indirect
185185
go.opentelemetry.io/otel v1.7.0 // indirect
186186
go.opentelemetry.io/otel/exporters/jaeger v1.7.0 // indirect

‎docs/examples/kubo-as-a-library/go.sum

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -198,6 +198,8 @@ github.com/edsrzf/mmap-go v1.0.0/go.mod h1:YO35OhQPt3KJa3ryjFM5Bs14WD66h8eGKpfaB
198198
github.com/elastic/gosigar v0.12.0/go.mod h1:iXRIGg2tLnu7LBdpqzyQfGDEidKCfWcCMS0WKyPWoMs=
199199
github.com/elastic/gosigar v0.14.2 h1:Dg80n8cr90OZ7x+bAax/QjoW/XqTI11RmA79ZwIm9/4=
200200
github.com/elastic/gosigar v0.14.2/go.mod h1:iXRIGg2tLnu7LBdpqzyQfGDEidKCfWcCMS0WKyPWoMs=
201+
github.com/emirpasic/gods v1.18.1 h1:FXtiHYKDGKCW2KzwZKx0iC0PQmdlorYgdFG9jPXJ1Bc=
202+
github.com/emirpasic/gods v1.18.1/go.mod h1:8tpGGwCnJ5H4r6BWwaV6OrWmMoPhUl5jm/FMNAnJvWQ=
201203
github.com/envoyproxy/go-control-plane v0.6.9/go.mod h1:SBwIajubJHhxtWwsL9s8ss4safvEdbitLhGGK48rN6g=
202204
github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
203205
github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
@@ -774,8 +776,8 @@ github.com/libp2p/go-libp2p-peerstore v0.2.2/go.mod h1:NQxhNjWxf1d4w6PihR8btWIRj
774776
github.com/libp2p/go-libp2p-peerstore v0.2.6/go.mod h1:ss/TWTgHZTMpsU/oKVVPQCGuDHItOpf2W8RxAi50P2s=
775777
github.com/libp2p/go-libp2p-peerstore v0.2.7/go.mod h1:ss/TWTgHZTMpsU/oKVVPQCGuDHItOpf2W8RxAi50P2s=
776778
github.com/libp2p/go-libp2p-pnet v0.2.0/go.mod h1:Qqvq6JH/oMZGwqs3N1Fqhv8NVhrdYcO0BW4wssv21LA=
777-
github.com/libp2p/go-libp2p-pubsub v0.8.2 h1:QLGUmkgKmwEVxVDYGsqc5t9CykOMY2Y21cXQHjR462I=
778-
github.com/libp2p/go-libp2p-pubsub v0.8.2/go.mod h1:e4kT+DYjzPUYGZeWk4I+oxCSYTXizzXii5LDRRhjKSw=
779+
github.com/libp2p/go-libp2p-pubsub v0.8.3 h1:T4+pcfcFm1K2v5oFyk68peSjVroaoM8zFygf6Y5WOww=
780+
github.com/libp2p/go-libp2p-pubsub v0.8.3/go.mod h1:eje970FXxjhtFbVEoiae+VUw24ZoSlk67BsiZPLRzlw=
779781
github.com/libp2p/go-libp2p-pubsub-router v0.6.0 h1:D30iKdlqDt5ZmLEYhHELCMRj8b4sFAqrUcshIUvVP/s=
780782
github.com/libp2p/go-libp2p-pubsub-router v0.6.0/go.mod h1:FY/q0/RBTKsLA7l4vqC2cbRbOvyDotg8PJQ7j8FDudE=
781783
github.com/libp2p/go-libp2p-quic-transport v0.10.0/go.mod h1:RfJbZ8IqXIhxBRm5hqUEJqjiiY8xmEuq3HUDS993MkA=
@@ -1285,8 +1287,6 @@ github.com/whyrusleeping/mdns v0.0.0-20180901202407-ef14215e6b30/go.mod h1:j4l84
12851287
github.com/whyrusleeping/mdns v0.0.0-20190826153040-b9b60ed33aa9/go.mod h1:j4l84WPFclQPj320J9gp0XwNKBb3U0zt5CBqjPp22G4=
12861288
github.com/whyrusleeping/multiaddr-filter v0.0.0-20160516205228-e903e4adabd7 h1:E9S12nwJwEOXe2d6gT6qxdvqMnNq+VnSsKPgm2ZZNds=
12871289
github.com/whyrusleeping/multiaddr-filter v0.0.0-20160516205228-e903e4adabd7/go.mod h1:X2c0RVCI1eSUFI8eLcY3c0423ykwiUdxLJtkDvruhjI=
1288-
github.com/whyrusleeping/timecache v0.0.0-20160911033111-cfcb2f1abfee h1:lYbXeSvJi5zk5GLKVuid9TVjS9a0OmLIDKTfoZBL6Ow=
1289-
github.com/whyrusleeping/timecache v0.0.0-20160911033111-cfcb2f1abfee/go.mod h1:m2aV4LZI4Aez7dP5PMyVKEHhUyEJ/RjmPEDOpDvudHg=
12901290
github.com/x-cray/logrus-prefixed-formatter v0.5.2/go.mod h1:2duySbKsL6M18s5GU7VPsoEPHyzalCE06qoARUCeBBE=
12911291
github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
12921292
github.com/xdg-go/scram v1.0.2/go.mod h1:1WAq6h33pAW+iRreB34OORO2Nf7qel3VV3fjBj+hCSs=

‎go.mod

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -74,7 +74,7 @@ require (
7474
github.com/libp2p/go-libp2p-http v0.4.0
7575
github.com/libp2p/go-libp2p-kad-dht v0.20.0
7676
github.com/libp2p/go-libp2p-kbucket v0.5.0
77-
github.com/libp2p/go-libp2p-pubsub v0.8.2
77+
github.com/libp2p/go-libp2p-pubsub v0.8.3
7878
github.com/libp2p/go-libp2p-pubsub-router v0.6.0
7979
github.com/libp2p/go-libp2p-record v0.2.0
8080
github.com/libp2p/go-libp2p-routing-helpers v0.6.0
@@ -132,6 +132,7 @@ require (
132132
github.com/dgraph-io/ristretto v0.0.2 // indirect
133133
github.com/docker/go-units v0.5.0 // indirect
134134
github.com/elastic/gosigar v0.14.2 // indirect
135+
github.com/emirpasic/gods v1.18.1 // indirect
135136
github.com/felixge/httpsnoop v1.0.2 // indirect
136137
github.com/flynn/noise v1.0.0 // indirect
137138
github.com/francoispqt/gojay v1.2.13 // indirect
@@ -223,7 +224,6 @@ require (
223224
github.com/whyrusleeping/cbor-gen v0.0.0-20221220214510-0333c149dec0 // indirect
224225
github.com/whyrusleeping/chunker v0.0.0-20181014151217-fe64bd25879f // indirect
225226
github.com/whyrusleeping/go-keyspace v0.0.0-20160322163242-5b898ac5add1 // indirect
226-
github.com/whyrusleeping/timecache v0.0.0-20160911033111-cfcb2f1abfee // indirect
227227
go.opentelemetry.io/otel/exporters/otlp/internal/retry v1.7.0 // indirect
228228
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.7.0 // indirect
229229
go.opentelemetry.io/otel/metric v0.30.0 // indirect

‎go.sum

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -206,6 +206,8 @@ github.com/elastic/gosigar v0.14.2 h1:Dg80n8cr90OZ7x+bAax/QjoW/XqTI11RmA79ZwIm9/
206206
github.com/elastic/gosigar v0.14.2/go.mod h1:iXRIGg2tLnu7LBdpqzyQfGDEidKCfWcCMS0WKyPWoMs=
207207
github.com/elgris/jsondiff v0.0.0-20160530203242-765b5c24c302 h1:QV0ZrfBLpFc2KDk+a4LJefDczXnonRwrYrQJY/9L4dA=
208208
github.com/elgris/jsondiff v0.0.0-20160530203242-765b5c24c302/go.mod h1:qBlWZqWeVx9BjvqBsnC/8RUlAYpIFmPvgROcw0n1scE=
209+
github.com/emirpasic/gods v1.18.1 h1:FXtiHYKDGKCW2KzwZKx0iC0PQmdlorYgdFG9jPXJ1Bc=
210+
github.com/emirpasic/gods v1.18.1/go.mod h1:8tpGGwCnJ5H4r6BWwaV6OrWmMoPhUl5jm/FMNAnJvWQ=
209211
github.com/envoyproxy/go-control-plane v0.6.9/go.mod h1:SBwIajubJHhxtWwsL9s8ss4safvEdbitLhGGK48rN6g=
210212
github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
211213
github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
@@ -808,8 +810,8 @@ github.com/libp2p/go-libp2p-peerstore v0.2.2/go.mod h1:NQxhNjWxf1d4w6PihR8btWIRj
808810
github.com/libp2p/go-libp2p-peerstore v0.2.6/go.mod h1:ss/TWTgHZTMpsU/oKVVPQCGuDHItOpf2W8RxAi50P2s=
809811
github.com/libp2p/go-libp2p-peerstore v0.2.7/go.mod h1:ss/TWTgHZTMpsU/oKVVPQCGuDHItOpf2W8RxAi50P2s=
810812
github.com/libp2p/go-libp2p-pnet v0.2.0/go.mod h1:Qqvq6JH/oMZGwqs3N1Fqhv8NVhrdYcO0BW4wssv21LA=
811-
github.com/libp2p/go-libp2p-pubsub v0.8.2 h1:QLGUmkgKmwEVxVDYGsqc5t9CykOMY2Y21cXQHjR462I=
812-
github.com/libp2p/go-libp2p-pubsub v0.8.2/go.mod h1:e4kT+DYjzPUYGZeWk4I+oxCSYTXizzXii5LDRRhjKSw=
813+
github.com/libp2p/go-libp2p-pubsub v0.8.3 h1:T4+pcfcFm1K2v5oFyk68peSjVroaoM8zFygf6Y5WOww=
814+
github.com/libp2p/go-libp2p-pubsub v0.8.3/go.mod h1:eje970FXxjhtFbVEoiae+VUw24ZoSlk67BsiZPLRzlw=
813815
github.com/libp2p/go-libp2p-pubsub-router v0.6.0 h1:D30iKdlqDt5ZmLEYhHELCMRj8b4sFAqrUcshIUvVP/s=
814816
github.com/libp2p/go-libp2p-pubsub-router v0.6.0/go.mod h1:FY/q0/RBTKsLA7l4vqC2cbRbOvyDotg8PJQ7j8FDudE=
815817
github.com/libp2p/go-libp2p-quic-transport v0.10.0/go.mod h1:RfJbZ8IqXIhxBRm5hqUEJqjiiY8xmEuq3HUDS993MkA=
@@ -1344,8 +1346,6 @@ github.com/whyrusleeping/mdns v0.0.0-20180901202407-ef14215e6b30/go.mod h1:j4l84
13441346
github.com/whyrusleeping/mdns v0.0.0-20190826153040-b9b60ed33aa9/go.mod h1:j4l84WPFclQPj320J9gp0XwNKBb3U0zt5CBqjPp22G4=
13451347
github.com/whyrusleeping/multiaddr-filter v0.0.0-20160516205228-e903e4adabd7 h1:E9S12nwJwEOXe2d6gT6qxdvqMnNq+VnSsKPgm2ZZNds=
13461348
github.com/whyrusleeping/multiaddr-filter v0.0.0-20160516205228-e903e4adabd7/go.mod h1:X2c0RVCI1eSUFI8eLcY3c0423ykwiUdxLJtkDvruhjI=
1347-
github.com/whyrusleeping/timecache v0.0.0-20160911033111-cfcb2f1abfee h1:lYbXeSvJi5zk5GLKVuid9TVjS9a0OmLIDKTfoZBL6Ow=
1348-
github.com/whyrusleeping/timecache v0.0.0-20160911033111-cfcb2f1abfee/go.mod h1:m2aV4LZI4Aez7dP5PMyVKEHhUyEJ/RjmPEDOpDvudHg=
13491349
github.com/x-cray/logrus-prefixed-formatter v0.5.2/go.mod h1:2duySbKsL6M18s5GU7VPsoEPHyzalCE06qoARUCeBBE=
13501350
github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
13511351
github.com/xdg-go/scram v1.0.2/go.mod h1:1WAq6h33pAW+iRreB34OORO2Nf7qel3VV3fjBj+hCSs=

0 commit comments

Comments
 (0)