Refactor GossipSub Construction (#1)

* Enables non-atomic validation for peer scoring parameters (libp2p#499)

* decouples topic scoring parameters

* adds skiping atomic validation for topic parameters

* cleans up

* adds skip atomic validation to peer score threshold

* adds skip atomic validation for peer parameters

* adds test for non-atomic validation

* adds tests for peer score

* adds tests for peer score thresholds

* refactors tests

* chore: Update .github/workflows/stale.yml [skip ci]

* adds with gossipsub tracker

Co-authored-by: libp2p-mgmt-read-write[bot] <104492852+libp2p-mgmt-read-write[bot]@users.noreply.github.com>

.github/workflows/stale.yml

@@ -0,0 +1,26 @@
+name: Close and mark stale issue
+
+on:
+  schedule:
+  - cron: '0 0 * * *'
+
+jobs:
+  stale:
+
+    runs-on: ubuntu-latest
+    permissions:
+      issues: write
+      pull-requests: write
+
+    steps:
+    - uses: actions/stale@v3
+      with:
+        repo-token: ${{ secrets.GITHUB_TOKEN }}
+        stale-issue-message: 'Oops, seems like we needed more information for this issue, please comment with more details or this issue will be closed in 7 days.'
+        close-issue-message: 'This issue was closed because it is missing author input.'
+        stale-issue-label: 'kind/stale'
+        any-of-labels: 'need/author-input'
+        exempt-issue-labels: 'need/triage,need/community-input,need/maintainer-input,need/maintainers-input,need/analysis,status/blocked,status/in-progress,status/ready,status/deferred,status/inactive'
+        days-before-issue-stale: 6
+        days-before-issue-close: 7
+        enable-statistics: true

gossipsub.go

@@ -204,10 +204,22 @@ type GossipSubParams struct {
 	IWantFollowupTime time.Duration
 }
 
-// NewGossipSub returns a new PubSub object using GossipSubRouter as the router.
+// NewGossipSub returns a new PubSub object using the default GossipSubRouter as the router.
 func NewGossipSub(ctx context.Context, h host.Host, opts ...Option) (*PubSub, error) {
+	rt := DefaultGossipSubRouter(h)
+	opts = append(opts, WithRawTracer(rt.tagTracer))
+	return NewGossipSubWithRouter(ctx, h, rt, opts...)
+}
+
+// NewGossipSubWithRouter returns a new PubSub object using the given router.
+func NewGossipSubWithRouter(ctx context.Context, h host.Host, rt PubSubRouter, opts ...Option) (*PubSub, error) {
+	return NewPubSub(ctx, h, rt, opts...)
+}
+
+// DefaultGossipSubRouter returns a new GossipSubRouter with default parameters.
+func DefaultGossipSubRouter(h host.Host) *GossipSubRouter {
 	params := DefaultGossipSubParams()
-	rt := &GossipSubRouter{
+	return &GossipSubRouter{
 		peers:     make(map[peer.ID]protocol.ID),
 		mesh:      make(map[string]map[peer.ID]struct{}),
 		fanout:    make(map[string]map[peer.ID]struct{}),
@@ -225,10 +237,6 @@ func NewGossipSub(ctx context.Context, h host.Host, opts ...Option) (*PubSub, er
 		tagTracer: newTagTracer(h.ConnManager()),
 		params:    params,
 	}
-
-	// hook the tag tracer
-	opts = append(opts, WithRawTracer(rt.tagTracer))
-	return NewPubSub(ctx, h, rt, opts...)
 }
 
 // DefaultGossipSubParams returns the default gossip sub parameters
@@ -1909,6 +1917,10 @@ func (gs *GossipSubRouter) getPeers(topic string, count int, filter func(peer.ID
 	return peers
 }
 
+func (gs *GossipSubRouter) WithTagTracerPubsubOption() Option {
+	return WithRawTracer(gs.tagTracer)
+}
+
 func peerListToMap(peers []peer.ID) map[peer.ID]struct{} {
 	pmap := make(map[peer.ID]struct{})
 	for _, p := range peers {

score_params.go

@@ -10,16 +10,19 @@ import (
 )
 
 type PeerScoreThresholds struct {
-	// GossipThreshold is the score threshold below which gossip propagation is supressed;
+	// whether it is allowed to just set some params and not all of them.
+	SkipAtomicValidation bool
+
+	// GossipThreshold is the score threshold below which gossip propagation is suppressed;
 	// should be negative.
 	GossipThreshold float64
 
 	// PublishThreshold is the score threshold below which we shouldn't publish when using flood
 	// publishing (also applies to fanout and floodsub peers); should be negative and <= GossipThreshold.
 	PublishThreshold float64
 
-	// GraylistThreshold is the score threshold below which message processing is supressed altogether,
-	// implementing an effective graylist according to peer score; should be negative and <= PublisThreshold.
+	// GraylistThreshold is the score threshold below which message processing is suppressed altogether,
+	// implementing an effective gray list according to peer score; should be negative and <= PublishThreshold.
 	GraylistThreshold float64
 
 	// AcceptPXThreshold is the score threshold below which PX will be ignored; this should be positive
@@ -32,25 +35,38 @@ type PeerScoreThresholds struct {
 }
 
 func (p *PeerScoreThresholds) validate() error {
-	if p.GossipThreshold > 0 || isInvalidNumber(p.GossipThreshold) {
-		return fmt.Errorf("invalid gossip threshold; it must be <= 0 and a valid number")
-	}
-	if p.PublishThreshold > 0 || p.PublishThreshold > p.GossipThreshold || isInvalidNumber(p.PublishThreshold) {
-		return fmt.Errorf("invalid publish threshold; it must be <= 0 and <= gossip threshold and a valid number")
-	}
-	if p.GraylistThreshold > 0 || p.GraylistThreshold > p.PublishThreshold || isInvalidNumber(p.GraylistThreshold) {
-		return fmt.Errorf("invalid graylist threshold; it must be <= 0 and <= publish threshold and a valid number")
+
+	if !p.SkipAtomicValidation || p.PublishThreshold != 0 || p.GossipThreshold != 0 || p.GraylistThreshold != 0 {
+		if p.GossipThreshold > 0 || isInvalidNumber(p.GossipThreshold) {
+			return fmt.Errorf("invalid gossip threshold; it must be <= 0 and a valid number")
+		}
+		if p.PublishThreshold > 0 || p.PublishThreshold > p.GossipThreshold || isInvalidNumber(p.PublishThreshold) {
+			return fmt.Errorf("invalid publish threshold; it must be <= 0 and <= gossip threshold and a valid number")
+		}
+		if p.GraylistThreshold > 0 || p.GraylistThreshold > p.PublishThreshold || isInvalidNumber(p.GraylistThreshold) {
+			return fmt.Errorf("invalid graylist threshold; it must be <= 0 and <= publish threshold and a valid number")
+		}
 	}
-	if p.AcceptPXThreshold < 0 || isInvalidNumber(p.AcceptPXThreshold) {
-		return fmt.Errorf("invalid accept PX threshold; it must be >= 0 and a valid number")
+
+	if !p.SkipAtomicValidation || p.AcceptPXThreshold != 0 {
+		if p.AcceptPXThreshold < 0 || isInvalidNumber(p.AcceptPXThreshold) {
+			return fmt.Errorf("invalid accept PX threshold; it must be >= 0 and a valid number")
+		}
 	}
-	if p.OpportunisticGraftThreshold < 0 || isInvalidNumber(p.OpportunisticGraftThreshold) {
-		return fmt.Errorf("invalid opportunistic grafting threshold; it must be >= 0 and a valid number")
+
+	if !p.SkipAtomicValidation || p.OpportunisticGraftThreshold != 0 {
+		if p.OpportunisticGraftThreshold < 0 || isInvalidNumber(p.OpportunisticGraftThreshold) {
+			return fmt.Errorf("invalid opportunistic grafting threshold; it must be >= 0 and a valid number")
+		}
 	}
+
 	return nil
 }
 
 type PeerScoreParams struct {
+	// whether it is allowed to just set some params and not all of them.
+	SkipAtomicValidation bool
+
 	// Score parameters per topic.
 	Topics map[string]*TopicScoreParams
 
@@ -99,12 +115,15 @@ type PeerScoreParams struct {
 }
 
 type TopicScoreParams struct {
+	// whether it is allowed to just set some params and not all of them.
+	SkipAtomicValidation bool
+
 	// The weight of the topic.
 	TopicWeight float64
 
 	// P1: time in the mesh
-	// This is the time the peer has ben grafted in the mesh.
-	// The value of of the parameter is the time/TimeInMeshQuantum, capped by TimeInMeshCap
+	// This is the time the peer has been grafted in the mesh.
+	// The value of the parameter is the time/TimeInMeshQuantum, capped by TimeInMeshCap.
 	// The weight of the parameter MUST be positive (or zero to disable).
 	TimeInMeshWeight  float64
 	TimeInMeshQuantum time.Duration
@@ -124,7 +143,7 @@ type TopicScoreParams struct {
 	// when validation succeeds.
 	// This window accounts for the minimum time before a hostile mesh peer trying to game the score
 	// could replay back a valid message we just sent them.
-	// It effectively tracks first and near-first deliveries, ie a message seen from a mesh peer
+	// It effectively tracks first and near-first deliveries, i.e., a message seen from a mesh peer
 	// before we have forwarded it to them.
 	// The parameter has an associated counter, decaying with MeshMessageDeliveriesDecay.
 	// If the counter exceeds the threshold, its value is 0.
@@ -159,41 +178,55 @@ func (p *PeerScoreParams) validate() error {
 		}
 	}
 
-	// check that the topic score is 0 or something positive
-	if p.TopicScoreCap < 0 || isInvalidNumber(p.TopicScoreCap) {
-		return fmt.Errorf("invalid topic score cap; must be positive (or 0 for no cap) and a valid number")
+	if !p.SkipAtomicValidation || p.TopicScoreCap != 0 {
+		// check that the topic score is 0 or something positive
+		if p.TopicScoreCap < 0 || isInvalidNumber(p.TopicScoreCap) {
+			return fmt.Errorf("invalid topic score cap; must be positive (or 0 for no cap) and a valid number")
+		}
 	}
 
 	// check that we have an app specific score; the weight can be anything (but expected positive)
 	if p.AppSpecificScore == nil {
-		return fmt.Errorf("missing application specific score function")
+		if p.SkipAtomicValidation {
+			p.AppSpecificScore = func(p peer.ID) float64 {
+				return 0
+			}
+		} else {
+			return fmt.Errorf("missing application specific score function")
+		}
 	}
 
-	// check the IP colocation factor
-	if p.IPColocationFactorWeight > 0 || isInvalidNumber(p.IPColocationFactorWeight) {
-		return fmt.Errorf("invalid IPColocationFactorWeight; must be negative (or 0 to disable) and a valid number")
-	}
-	if p.IPColocationFactorWeight != 0 && p.IPColocationFactorThreshold < 1 {
-		return fmt.Errorf("invalid IPColocationFactorThreshold; must be at least 1")
+	if !p.SkipAtomicValidation || p.IPColocationFactorWeight != 0 {
+		// check the IP collocation factor
+		if p.IPColocationFactorWeight > 0 || isInvalidNumber(p.IPColocationFactorWeight) {
+			return fmt.Errorf("invalid IPColocationFactorWeight; must be negative (or 0 to disable) and a valid number")
+		}
+		if p.IPColocationFactorWeight != 0 && p.IPColocationFactorThreshold < 1 {
+			return fmt.Errorf("invalid IPColocationFactorThreshold; must be at least 1")
+		}
 	}
 
 	// check the behaviour penalty
-	if p.BehaviourPenaltyWeight > 0 || isInvalidNumber(p.BehaviourPenaltyWeight) {
-		return fmt.Errorf("invalid BehaviourPenaltyWeight; must be negative (or 0 to disable) and a valid number")
-	}
-	if p.BehaviourPenaltyWeight != 0 && (p.BehaviourPenaltyDecay <= 0 || p.BehaviourPenaltyDecay >= 1 || isInvalidNumber(p.BehaviourPenaltyDecay)) {
-		return fmt.Errorf("invalid BehaviourPenaltyDecay; must be between 0 and 1")
-	}
-	if p.BehaviourPenaltyThreshold < 0 || isInvalidNumber(p.BehaviourPenaltyThreshold) {
-		return fmt.Errorf("invalid BehaviourPenaltyThreshold; must be >= 0 and a valid number")
+	if !p.SkipAtomicValidation || p.BehaviourPenaltyWeight != 0 || p.BehaviourPenaltyThreshold != 0 {
+		if p.BehaviourPenaltyWeight > 0 || isInvalidNumber(p.BehaviourPenaltyWeight) {
+			return fmt.Errorf("invalid BehaviourPenaltyWeight; must be negative (or 0 to disable) and a valid number")
+		}
+		if p.BehaviourPenaltyWeight != 0 && (p.BehaviourPenaltyDecay <= 0 || p.BehaviourPenaltyDecay >= 1 || isInvalidNumber(p.BehaviourPenaltyDecay)) {
+			return fmt.Errorf("invalid BehaviourPenaltyDecay; must be between 0 and 1")
+		}
+		if p.BehaviourPenaltyThreshold < 0 || isInvalidNumber(p.BehaviourPenaltyThreshold) {
+			return fmt.Errorf("invalid BehaviourPenaltyThreshold; must be >= 0 and a valid number")
+		}
 	}
 
 	// check the decay parameters
-	if p.DecayInterval < time.Second {
-		return fmt.Errorf("invalid DecayInterval; must be at least 1s")
-	}
-	if p.DecayToZero <= 0 || p.DecayToZero >= 1 || isInvalidNumber(p.DecayToZero) {
-		return fmt.Errorf("invalid DecayToZero; must be between 0 and 1")
+	if !p.SkipAtomicValidation || p.DecayInterval != 0 || p.DecayToZero != 0 {
+		if p.DecayInterval < time.Second {
+			return fmt.Errorf("invalid DecayInterval; must be at least 1s")
+		}
+		if p.DecayToZero <= 0 || p.DecayToZero >= 1 || isInvalidNumber(p.DecayToZero) {
+			return fmt.Errorf("invalid DecayToZero; must be between 0 and 1")
+		}
 	}
 
 	// no need to check the score retention; a value of 0 means that we don't retain scores
@@ -207,6 +240,43 @@ func (p *TopicScoreParams) validate() error {
 	}
 
 	// check P1
+	if err := p.validateTimeInMeshParams(); err != nil {
+		return err
+	}
+
+	// check P2
+	if err := p.validateMessageDeliveryParams(); err != nil {
+		return err
+	}
+	// check P3
+	if err := p.validateMeshMessageDeliveryParams(); err != nil {
+		return err
+	}
+
+	// check P3b
+	if err := p.validateMessageFailurePenaltyParams(); err != nil {
+		return err
+	}
+
+	// check P4
+	if err := p.validateInvalidMessageDeliveryParams(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (p *TopicScoreParams) validateTimeInMeshParams() error {
+	if p.SkipAtomicValidation {
+		// in non-atomic mode, parameters at their zero values are dismissed from validation.
+		if p.TimeInMeshWeight == 0 && p.TimeInMeshQuantum == 0 && p.TimeInMeshCap == 0 {
+			return nil
+		}
+	}
+
+	// either atomic validation mode, or some parameters have been set a value,
+	// hence, proceed with normal validation of all related parameters in this context.
+
 	if p.TimeInMeshQuantum == 0 {
 		return fmt.Errorf("invalid TimeInMeshQuantum; must be non zero")
 	}
@@ -220,7 +290,20 @@ func (p *TopicScoreParams) validate() error {
 		return fmt.Errorf("invalid TimeInMeshCap; must be positive and a valid number")
 	}
 
-	// check P2
+	return nil
+}
+
+func (p *TopicScoreParams) validateMessageDeliveryParams() error {
+	if p.SkipAtomicValidation {
+		// in non-atomic mode, parameters at their zero values are dismissed from validation.
+		if p.FirstMessageDeliveriesWeight == 0 && p.FirstMessageDeliveriesCap == 0 && p.FirstMessageDeliveriesDecay == 0 {
+			return nil
+		}
+	}
+
+	// either atomic validation mode, or some parameters have been set a value,
+	// hence, proceed with normal validation of all related parameters in this context.
+
 	if p.FirstMessageDeliveriesWeight < 0 || isInvalidNumber(p.FirstMessageDeliveriesWeight) {
 		return fmt.Errorf("invallid FirstMessageDeliveriesWeight; must be positive (or 0 to disable) and a valid number")
 	}
@@ -231,7 +314,25 @@ func (p *TopicScoreParams) validate() error {
 		return fmt.Errorf("invalid FirstMessageDeliveriesCap; must be positive and a valid number")
 	}
 
-	// check P3
+	return nil
+}
+
+func (p *TopicScoreParams) validateMeshMessageDeliveryParams() error {
+	if p.SkipAtomicValidation {
+		// in non-atomic mode, parameters at their zero values are dismissed from validation.
+		if p.MeshMessageDeliveriesWeight == 0 &&
+			p.MeshMessageDeliveriesCap == 0 &&
+			p.MeshMessageDeliveriesDecay == 0 &&
+			p.MeshMessageDeliveriesThreshold == 0 &&
+			p.MeshMessageDeliveriesWindow == 0 &&
+			p.MeshMessageDeliveriesActivation == 0 {
+			return nil
+		}
+	}
+
+	// either atomic validation mode, or some parameters have been set a value,
+	// hence, proceed with normal validation of all related parameters in this context.
+
 	if p.MeshMessageDeliveriesWeight > 0 || isInvalidNumber(p.MeshMessageDeliveriesWeight) {
 		return fmt.Errorf("invalid MeshMessageDeliveriesWeight; must be negative (or 0 to disable) and a valid number")
 	}
@@ -251,15 +352,41 @@ func (p *TopicScoreParams) validate() error {
 		return fmt.Errorf("invalid MeshMessageDeliveriesActivation; must be at least 1s")
 	}
 
-	// check P3b
+	return nil
+}
+
+func (p *TopicScoreParams) validateMessageFailurePenaltyParams() error {
+	if p.SkipAtomicValidation {
+		// in selective mode, parameters at their zero values are dismissed from validation.
+		if p.MeshFailurePenaltyDecay == 0 && p.MeshFailurePenaltyWeight == 0 {
+			return nil
+		}
+	}
+
+	// either atomic validation mode, or some parameters have been set a value,
+	// hence, proceed with normal validation of all related parameters in this context.
+
 	if p.MeshFailurePenaltyWeight > 0 || isInvalidNumber(p.MeshFailurePenaltyWeight) {
 		return fmt.Errorf("invalid MeshFailurePenaltyWeight; must be negative (or 0 to disable) and a valid number")
 	}
 	if p.MeshFailurePenaltyWeight != 0 && (isInvalidNumber(p.MeshFailurePenaltyDecay) || p.MeshFailurePenaltyDecay <= 0 || p.MeshFailurePenaltyDecay >= 1) {
 		return fmt.Errorf("invalid MeshFailurePenaltyDecay; must be between 0 and 1")
 	}
 
-	// check P4
+	return nil
+}
+
+func (p *TopicScoreParams) validateInvalidMessageDeliveryParams() error {
+	if p.SkipAtomicValidation {
+		// in selective mode, parameters at their zero values are dismissed from validation.
+		if p.InvalidMessageDeliveriesDecay == 0 && p.InvalidMessageDeliveriesWeight == 0 {
+			return nil
+		}
+	}
+
+	// either atomic validation mode, or some parameters have been set a value,
+	// hence, proceed with normal validation of all related parameters in this context.
+
 	if p.InvalidMessageDeliveriesWeight > 0 || isInvalidNumber(p.InvalidMessageDeliveriesWeight) {
 		return fmt.Errorf("invalid InvalidMessageDeliveriesWeight; must be negative (or 0 to disable) and a valid number")
 	}
@@ -281,7 +408,7 @@ func ScoreParameterDecay(decay time.Duration) float64 {
 	return ScoreParameterDecayWithBase(decay, DefaultDecayInterval, DefaultDecayToZero)
 }
 
-// ScoreParameterDecay computes the decay factor for a parameter using base as the DecayInterval
+// ScoreParameterDecayWithBase computes the decay factor for a parameter using base as the DecayInterval
 func ScoreParameterDecayWithBase(decay time.Duration, base time.Duration, decayToZero float64) float64 {
 	// the decay is linear, so after n ticks the value is factor^n
 	// so factor^n = decayToZero => factor = decayToZero^(1/n)