feat: Pubsub.SeenMessagesStrategy (#9543)
* feat: expire messages from the cache based on last seen time * docs: Pubsub.SeenMessagesStrategy Ref. https://github.com/libp2p/go-libp2p-pubsub/pull/513 Co-authored-by: Marcin Rataj <lidel@lidel.org>
Mohsin Zaidi committed
Jan 26, 2023 at 18:24 UTC
9652f24f6c1dce639444b8bb39caa0a9cd375870
9 files changed
+191
-53
config/pubsub.go
+25
-2
@@ -1,5 +1,24 @@
1
package config
2
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
+
22
type PubsubConfig struct {
23
// Router can be either floodsub (legacy) or gossipsub (new and
24
// backwards compatible).
@@ -12,7 +31,11 @@ type PubsubConfig struct {
31
// Enable pubsub (--enable-pubsub-experiment)
32
Enabled Flag `json:",omitempty"`
33
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.
36
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"`
41
}
core/node/groups.go
+13
@@ -11,6 +11,7 @@ import (
11
"github.com/ipfs/go-log"
12
"github.com/ipfs/kubo/config"
13
pubsub "github.com/libp2p/go-libp2p-pubsub"
14
+ "github.com/libp2p/go-libp2p-pubsub/timecache"
15
"github.com/libp2p/go-libp2p/core/peer"
16
17
"github.com/ipfs/kubo/core/node/libp2p"
@@ -66,6 +67,18 @@ func LibP2P(bcfg *BuildCfg, cfg *config.Config) fx.Option {
67
pubsub.WithSeenMessagesTTL(cfg.Pubsub.SeenMessagesTTL.WithDefault(pubsub.TimeCacheDuration)),
68
)
69
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
+
82
switch cfg.Pubsub.Router {
83
case "":
84
fallthrough
docs/changelogs/v0.18.md
+38
@@ -1,5 +1,43 @@
1
# Kubo changelog v0.18
2
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
+
41
## v0.18.0
42
43
### Overview
docs/config.md
+26
-2
@@ -100,6 +100,7 @@ config file at runtime.
100
- [`Pubsub.Router`](#pubsubrouter)
101
- [`Pubsub.DisableSigning`](#pubsubdisablesigning)
102
- [`Pubsub.SeenMessagesTTL`](#pubsubseenmessagesttl)
103
+ - [`Pubsub.SeenMessagesStrategy`](#pubsubseenmessagesstrategy)
104
- [`Peering`](#peering)
105
- [`Peering.Peers`](#peeringpeers)
106
- [`Reprovider`](#reprovider)
@@ -1206,8 +1207,8 @@ Type: `bool`
1207
1208
### `Pubsub.SeenMessagesTTL`
1209
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.
1212
1213
A smaller value for this parameter means that Pubsub messages in the cache will
1214
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
1224
1225
Type: `optionalDuration`
1226
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
+
1250
## `Peering`
1251
1252
Configures the peering subsystem. The peering subsystem configures Kubo to
docs/examples/kubo-as-a-library/go.mod
+2
-2
@@ -38,6 +38,7 @@ require (
38
github.com/docker/go-units v0.5.0 // indirect
39
github.com/dustin/go-humanize v1.0.0 // indirect
40
github.com/elastic/gosigar v0.14.2 // indirect
41
+ github.com/emirpasic/gods v1.18.1 // indirect
42
github.com/facebookgo/atomicfile v0.0.0-20151019160806-2de1f203e7d5 // indirect
43
github.com/flynn/noise v1.0.0 // indirect
44
github.com/francoispqt/gojay v1.2.13 // indirect
@@ -122,7 +123,7 @@ require (
123
github.com/libp2p/go-libp2p-asn-util v0.2.0 // indirect
124
github.com/libp2p/go-libp2p-kad-dht v0.20.0 // indirect
125
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
127
github.com/libp2p/go-libp2p-pubsub-router v0.6.0 // indirect
128
github.com/libp2p/go-libp2p-record v0.2.0 // indirect
129
github.com/libp2p/go-libp2p-routing-helpers v0.6.0 // indirect
@@ -180,7 +181,6 @@ require (
181
github.com/whyrusleeping/chunker v0.0.0-20181014151217-fe64bd25879f // indirect
182
github.com/whyrusleeping/go-keyspace v0.0.0-20160322163242-5b898ac5add1 // indirect
183
github.com/whyrusleeping/multiaddr-filter v0.0.0-20160516205228-e903e4adabd7 // indirect
183
- github.com/whyrusleeping/timecache v0.0.0-20160911033111-cfcb2f1abfee // indirect
184
go.opencensus.io v0.24.0 // indirect
185
go.opentelemetry.io/otel v1.7.0 // indirect
186
go.opentelemetry.io/otel/exporters/jaeger v1.7.0 // indirect
docs/examples/kubo-as-a-library/go.sum
+4
-4
@@ -198,6 +198,8 @@ github.com/edsrzf/mmap-go v1.0.0/go.mod h1:YO35OhQPt3KJa3ryjFM5Bs14WD66h8eGKpfaB
198
github.com/elastic/gosigar v0.12.0/go.mod h1:iXRIGg2tLnu7LBdpqzyQfGDEidKCfWcCMS0WKyPWoMs=
199
github.com/elastic/gosigar v0.14.2 h1:Dg80n8cr90OZ7x+bAax/QjoW/XqTI11RmA79ZwIm9/4=
200
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=
203
github.com/envoyproxy/go-control-plane v0.6.9/go.mod h1:SBwIajubJHhxtWwsL9s8ss4safvEdbitLhGGK48rN6g=
204
github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
205
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
776
github.com/libp2p/go-libp2p-peerstore v0.2.6/go.mod h1:ss/TWTgHZTMpsU/oKVVPQCGuDHItOpf2W8RxAi50P2s=
777
github.com/libp2p/go-libp2p-peerstore v0.2.7/go.mod h1:ss/TWTgHZTMpsU/oKVVPQCGuDHItOpf2W8RxAi50P2s=
778
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=
781
github.com/libp2p/go-libp2p-pubsub-router v0.6.0 h1:D30iKdlqDt5ZmLEYhHELCMRj8b4sFAqrUcshIUvVP/s=
782
github.com/libp2p/go-libp2p-pubsub-router v0.6.0/go.mod h1:FY/q0/RBTKsLA7l4vqC2cbRbOvyDotg8PJQ7j8FDudE=
783
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
1287
github.com/whyrusleeping/mdns v0.0.0-20190826153040-b9b60ed33aa9/go.mod h1:j4l84WPFclQPj320J9gp0XwNKBb3U0zt5CBqjPp22G4=
1288
github.com/whyrusleeping/multiaddr-filter v0.0.0-20160516205228-e903e4adabd7 h1:E9S12nwJwEOXe2d6gT6qxdvqMnNq+VnSsKPgm2ZZNds=
1289
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=
1290
github.com/x-cray/logrus-prefixed-formatter v0.5.2/go.mod h1:2duySbKsL6M18s5GU7VPsoEPHyzalCE06qoARUCeBBE=
1291
github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
1292
github.com/xdg-go/scram v1.0.2/go.mod h1:1WAq6h33pAW+iRreB34OORO2Nf7qel3VV3fjBj+hCSs=
go.mod
+2
-2
@@ -74,7 +74,7 @@ require (
74
github.com/libp2p/go-libp2p-http v0.4.0
75
github.com/libp2p/go-libp2p-kad-dht v0.20.0
76
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
78
github.com/libp2p/go-libp2p-pubsub-router v0.6.0
79
github.com/libp2p/go-libp2p-record v0.2.0
80
github.com/libp2p/go-libp2p-routing-helpers v0.6.0
@@ -132,6 +132,7 @@ require (
132
github.com/dgraph-io/ristretto v0.0.2 // indirect
133
github.com/docker/go-units v0.5.0 // indirect
134
github.com/elastic/gosigar v0.14.2 // indirect
135
+ github.com/emirpasic/gods v1.18.1 // indirect
136
github.com/felixge/httpsnoop v1.0.2 // indirect
137
github.com/flynn/noise v1.0.0 // indirect
138
github.com/francoispqt/gojay v1.2.13 // indirect
@@ -223,7 +224,6 @@ require (
224
github.com/whyrusleeping/cbor-gen v0.0.0-20221220214510-0333c149dec0 // indirect
225
github.com/whyrusleeping/chunker v0.0.0-20181014151217-fe64bd25879f // indirect
226
github.com/whyrusleeping/go-keyspace v0.0.0-20160322163242-5b898ac5add1 // indirect
226
- github.com/whyrusleeping/timecache v0.0.0-20160911033111-cfcb2f1abfee // indirect
227
go.opentelemetry.io/otel/exporters/otlp/internal/retry v1.7.0 // indirect
228
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.7.0 // indirect
229
go.opentelemetry.io/otel/metric v0.30.0 // indirect
go.sum
+4
-4
@@ -206,6 +206,8 @@ github.com/elastic/gosigar v0.14.2 h1:Dg80n8cr90OZ7x+bAax/QjoW/XqTI11RmA79ZwIm9/
206
github.com/elastic/gosigar v0.14.2/go.mod h1:iXRIGg2tLnu7LBdpqzyQfGDEidKCfWcCMS0WKyPWoMs=
207
github.com/elgris/jsondiff v0.0.0-20160530203242-765b5c24c302 h1:QV0ZrfBLpFc2KDk+a4LJefDczXnonRwrYrQJY/9L4dA=
208
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=
211
github.com/envoyproxy/go-control-plane v0.6.9/go.mod h1:SBwIajubJHhxtWwsL9s8ss4safvEdbitLhGGK48rN6g=
212
github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
213
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
810
github.com/libp2p/go-libp2p-peerstore v0.2.6/go.mod h1:ss/TWTgHZTMpsU/oKVVPQCGuDHItOpf2W8RxAi50P2s=
811
github.com/libp2p/go-libp2p-peerstore v0.2.7/go.mod h1:ss/TWTgHZTMpsU/oKVVPQCGuDHItOpf2W8RxAi50P2s=
812
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=
815
github.com/libp2p/go-libp2p-pubsub-router v0.6.0 h1:D30iKdlqDt5ZmLEYhHELCMRj8b4sFAqrUcshIUvVP/s=
816
github.com/libp2p/go-libp2p-pubsub-router v0.6.0/go.mod h1:FY/q0/RBTKsLA7l4vqC2cbRbOvyDotg8PJQ7j8FDudE=
817
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
1346
github.com/whyrusleeping/mdns v0.0.0-20190826153040-b9b60ed33aa9/go.mod h1:j4l84WPFclQPj320J9gp0XwNKBb3U0zt5CBqjPp22G4=
1347
github.com/whyrusleeping/multiaddr-filter v0.0.0-20160516205228-e903e4adabd7 h1:E9S12nwJwEOXe2d6gT6qxdvqMnNq+VnSsKPgm2ZZNds=
1348
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=
1349
github.com/x-cray/logrus-prefixed-formatter v0.5.2/go.mod h1:2duySbKsL6M18s5GU7VPsoEPHyzalCE06qoARUCeBBE=
1350
github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
1351
github.com/xdg-go/scram v1.0.2/go.mod h1:1WAq6h33pAW+iRreB34OORO2Nf7qel3VV3fjBj+hCSs=
test/integration/pubsub_msg_seen_cache_test.go
+77
-37
@@ -22,6 +22,7 @@ import (
22
23
"github.com/libp2p/go-libp2p-pubsub"
24
"github.com/libp2p/go-libp2p-pubsub/pb"
25
+ "github.com/libp2p/go-libp2p-pubsub/timecache"
26
"github.com/libp2p/go-libp2p/core/peer"
27
28
mock "github.com/ipfs/kubo/core/mock"
@@ -76,7 +77,6 @@ func RunMessageSeenCacheTTLTest(t *testing.T, seenMessagesCacheTTL string) error
77
78
var bootstrapNode, consumerNode, producerNode *core.IpfsNode
79
var bootstrapPeerID, consumerPeerID, producerPeerID peer.ID
79
- sendDupMsg := false
80
81
mn := mocknet.New()
82
bootstrapNode, err := mockNode(ctx, mn, false, "") // no need for PubSub configuration
@@ -98,6 +98,12 @@ func RunMessageSeenCacheTTLTest(t *testing.T, seenMessagesCacheTTL string) error
98
t.Fatal(err)
99
}
100
101
+ // Used for logging the timeline
102
+ startTime := time.Time{}
103
+
104
+ // Used for overriding the message ID
105
+ sendMsgID := ""
106
+
107
// Set up the pubsub message ID generation override for the producer
108
core.RegisterFXOptionFunc(func(info core.FXNodeInfo) ([]fx.Option, error) {
109
var pubsubOptions []pubsub.Option
@@ -105,19 +111,23 @@ func RunMessageSeenCacheTTLTest(t *testing.T, seenMessagesCacheTTL string) error
111
pubsubOptions,
112
pubsub.WithSeenMessagesTTL(ttl),
113
pubsub.WithMessageIdFn(func(pmsg *pubsub_pb.Message) string {
108
- now := time.Now().Format(time.StampMilli)
114
+ now := time.Now()
115
+ if startTime.Second() == 0 {
116
+ startTime = now
117
+ }
118
+ timeElapsed := now.Sub(startTime).Seconds()
119
msg := string(pmsg.Data)
110
- var msgID string
120
from, _ := peer.IDFromBytes(pmsg.From)
112
- if (from == producerPeerID) && sendDupMsg {
113
- msgID = "DupMsg"
114
- t.Logf("sending [%s] with duplicate message ID at [%s]", msg, now)
121
+ var msgID string
122
+ if from == producerPeerID {
123
+ msgID = sendMsgID
124
+ t.Logf("sending [%s] with message ID [%s] at T%fs", msg, msgID, timeElapsed)
125
} else {
126
msgID = pubsub.DefaultMsgIdFn(pmsg)
117
- t.Logf("sending [%s] with unique message ID at [%s]", msg, now)
127
}
128
return msgID
129
}),
130
+ pubsub.WithSeenMessagesStrategy(timecache.Strategy_LastSeen),
131
)
132
return append(
133
info.FXOptions,
@@ -165,8 +175,8 @@ func RunMessageSeenCacheTTLTest(t *testing.T, seenMessagesCacheTTL string) error
175
t.Fatal(err)
176
}
177
// Utility functions defined inline to include context in closure
168
- now := func() string {
169
- return time.Now().Format(time.StampMilli)
178
+ now := func() float64 {
179
+ return time.Since(startTime).Seconds()
180
}
181
ctr := 0
182
msgGen := func() string {
@@ -188,57 +198,87 @@ func RunMessageSeenCacheTTLTest(t *testing.T, seenMessagesCacheTTL string) error
198
msg, err := consumerSubscription.Next(rxCtx)
199
if shouldFind {
200
if err != nil {
191
- t.Logf("did not receive [%s] by [%s]", msgTxt, now())
201
+ t.Logf("expected but did not receive [%s] at T%fs", msgTxt, now())
202
t.Fatal(err)
203
}
194
- t.Logf("received [%s] at [%s]", string(msg.Data()), now())
204
+ t.Logf("received [%s] at T%fs", string(msg.Data()), now())
205
if !bytes.Equal(msg.Data(), []byte(msgTxt)) {
206
t.Fatalf("consumed data [%s] does not match published data [%s]", string(msg.Data()), msgTxt)
207
}
208
} else {
209
if err == nil {
200
- t.Logf("received [%s] at [%s]", string(msg.Data()), now())
210
+ t.Logf("not expected but received [%s] at T%fs", string(msg.Data()), now())
211
t.Fail()
212
}
203
- t.Logf("did not receive [%s] by [%s]", msgTxt, now())
213
+ t.Logf("did not receive [%s] at T%fs", msgTxt, now())
214
}
215
}
216
207
- // Send message 1 with the message ID we're going to duplicate later
208
- sendDupMsg = true
217
+ const MsgID1 = "MsgID1"
218
+ const MsgID2 = "MsgID2"
219
+ const MsgID3 = "MsgID3"
220
+
221
+ // Send message 1 with the message ID we're going to duplicate
222
+ sentMsg1 := time.Now()
223
+ sendMsgID = MsgID1
224
msgTxt := produceMessage()
210
- consumeMessage(msgTxt, true) // should find message
225
+ // Should find the message because it's new
226
+ consumeMessage(msgTxt, true)
227
212
- // Send message 2 with the same message ID as before
213
- sendDupMsg = true
228
+ // Send message 2 with a duplicate message ID
229
+ sendMsgID = MsgID1
230
msgTxt = produceMessage()
215
- consumeMessage(msgTxt, false) // should NOT find message, because it got deduplicated (sent twice within the SeenMessagesTTL window)
216
-
217
- // Wait for seen cache TTL time to let seen cache entries time out
218
- time.Sleep(ttl)
231
+ // Should NOT find message because it got deduplicated (sent 2 times within the SeenMessagesTTL window).
232
+ consumeMessage(msgTxt, false)
233
234
// Send message 3 with a new message ID
221
- //
222
- // This extra step is necessary for testing the cache TTL because the PubSub code only garbage collects when a
223
- // message ID was not already present in the cache. This means that message 2's cache entry, even though it has
224
- // technically timed out, will still cause the message to be considered duplicate. When a message with a different
225
- // ID passes through, it will be added to the cache and garbage collection will clean up message 2's entry. This is
226
- // another bug in the pubsub/cache implementation that will be fixed once the code is refactored for this issue:
227
- // https://github.com/libp2p/go-libp2p-pubsub/issues/502
228
- sendDupMsg = false
235
+ sendMsgID = MsgID2
236
msgTxt = produceMessage()
230
- consumeMessage(msgTxt, true) // should find message
237
+ // Should find the message because it's new
238
+ consumeMessage(msgTxt, true)
239
+
240
+ // Wait till just before the SeenMessagesTTL window has passed since message 1 was sent
241
+ time.Sleep(time.Until(sentMsg1.Add(ttl - 100*time.Millisecond)))
242
+
243
+ // Send message 4 with a duplicate message ID
244
+ sendMsgID = MsgID1
245
+ msgTxt = produceMessage()
246
+ // Should NOT find the message because it got deduplicated (sent 3 times within the SeenMessagesTTL window). This
247
+ // time, however, the expiration for the message should also get pushed out for a whole SeenMessagesTTL window since
248
+ // the default time cache now implements a sliding window algorithm.
249
+ consumeMessage(msgTxt, false)
250
+
251
+ // Send message 5 with a duplicate message ID. This will be a second after the last attempt above since NOT finding
252
+ // a message takes a second to determine. That would put this attempt at ~1 second after the SeenMessagesTTL window
253
+ // starting at message 1 has expired.
254
+ sentMsg5 := time.Now()
255
+ sendMsgID = MsgID1
256
+ msgTxt = produceMessage()
257
+ // Should NOT find the message, because it got deduplicated (sent 2 times since the updated SeenMessagesTTL window
258
+ // started). This time again, the expiration should get pushed out for another SeenMessagesTTL window.
259
+ consumeMessage(msgTxt, false)
260
+
261
+ // Send message 6 with a message ID that hasn't been seen within a SeenMessagesTTL window
262
+ sendMsgID = MsgID2
263
+ msgTxt = produceMessage()
264
+ // Should find the message since last read > SeenMessagesTTL, so it looks like a new message.
265
+ consumeMessage(msgTxt, true)
266
+
267
+ // Sleep for a full SeenMessagesTTL window to let cache entries time out
268
+ time.Sleep(time.Until(sentMsg5.Add(ttl + 100*time.Millisecond)))
269
232
- // Send message 4 with the same message ID as before
233
- sendDupMsg = true
270
+ // Send message 7 with a duplicate message ID
271
+ sendMsgID = MsgID1
272
msgTxt = produceMessage()
235
- consumeMessage(msgTxt, true) // should find message again (time since the last read > SeenMessagesTTL, so it looks like a new message).
273
+ // Should find the message this time since last read > SeenMessagesTTL, so it looks like a new message.
274
+ consumeMessage(msgTxt, true)
275
237
- // Send message 5 with a new message ID
276
+ // Send message 8 with a brand new message ID
277
//
278
// This step is not strictly necessary, but has been added for good measure.
240
- sendDupMsg = false
279
+ sendMsgID = MsgID3
280
msgTxt = produceMessage()
242
- consumeMessage(msgTxt, true) // should find message
281
+ // Should find the message because it's new
282
+ consumeMessage(msgTxt, true)
283
return nil
284
}