go-libp2p-pubsub

The Topic type is the primary handle for interacting with a specific pubsub topic. It allows you to publish messages, subscribe to incoming messages, manage peer events, and configure topic-specific settings like peer scoring.

Common tasks include:

• Publishing: Use Publish to send data to the topic.
• Subscribing: Use Subscribe to receive a stream of messages.
• Monitoring Peers: Use EventHandler to track peers joining or leaving the topic.
• Relaying: Use Relay to enable message relaying for the topic.
• Closing: Use Close to shut down the topic handle.

// Example of basic topic usage
// (Assuming a PubSub instance 'ps' is already initialized)
topic, err := ps.Join("my-topic")
if err != nil {
    return err
}

// Publish a message
err = topic.Publish(ctx, []byte("hello world"))

// Subscribe to messages
sub, err := topic.Subscribe()
if err != nil {
    return err
}

// Read from subscription
msg, err := sub.Next(ctx)
if err != nil {
    return err
}
fmt.Printf("Received: %s\n", string(msg.Data))

The gossipTracer is an internal implementation of the RawTracer interface used to track IWANT requests. It monitors whether peers follow up on IWANT requests after an IHAVE advertisement.

To prevent excessive memory usage, the tracer uses probabilistic tracking of promises. It identifies 'broken promises'—instances where a peer failed to deliver a requested message within the configured followUpTime—which can then be used to penalize those peers.

Key lifecycle methods for the tracer include:

• DeliverMessage(msg *Message): Fulfills a promise when a message is successfully delivered.
• RejectMessage(msg *Message, reason string): Fulfills a promise when a message is rejected (except for specific signature-related reasons), allowing the score penalty to be applied based on the invalid delivery.
• ValidateMessage(msg *Message): Fulfills a promise as soon as message validation begins.
• ThrottlePeer(p peer.ID): Immediately voids all outstanding promises associated with a specific peer, typically called when a peer is being throttled.

The MessageSignaturePolicy type defines how a pubsub node handles message signatures. It determines whether the node produces signatures for its own messages and whether it expects and verifies signatures on incoming messages.

There are two primary modes of operation:

1. Strict Mode: Use StrictSign to ensure all outgoing messages are signed and all incoming messages are verified. Use StrictNoSign to disable signatures entirely; in this mode, the node will drop and penalize incoming messages that attempt to carry a signature.
2. Lax Mode (Deprecated): LaxSign and LaxNoSign are deprecated. It is recommended to use either strict signing or strict no-signing to avoid ambiguity.

Note: StrictNoSign is defined as msgVerification in the source, which implies it enforces the verification logic (checking for the absence of signatures) rather than just ignoring them.

const (
	// StrictSign produces signatures and expects and verifies incoming signatures
	StrictSign = msgSigning | msgVerification
	// StrictNoSign does not produce signatures and drops and penalises incoming messages that carry one
	StrictNoSign = msgVerification
)

// Deprecated: it is recommend to either strictly enable, or strictly disable, signatures.
const (
	LaxSign = msgSigning
	LaxNoSign = 0
)

A SubscriptionFilter is used to decide which topics a node is interested in. It controls both local subscription attempts and incoming subscription notifications from other peers.

When a SubscriptionFilter is applied:

1. Joining Topics: If CanSubscribe(topic) returns false, the local Join operation will fail with an error.
2. Incoming Notifications: When a peer notifies you of its subscriptions via RPC, FilterIncomingSubscriptions is called. If it returns false or filters out a topic, that notification is ignored.

You can apply a filter to your PubSub instance using the WithSubscriptionFilter option during initialization.

import (
	"github.com/libp2p/go-libp2p-pubsub"
	"github.com/libp2p/go-libp2p"
)

// Example: Creating a PubSub instance with an allowlist filter
filter := pubsub.NewAllowlistSubscriptionFilter("topic1", "topic2")
ps, err := pubsub.New("noise", pubsub.WithSubscriptionFilter(filter))

You can enable periodic inspection of peer scores in a GossipSub router using the WithPeerScoreInspect option. This is useful for debugging how peers are being scored and identifying potential issues with peer behavior or scoring parameters.

There are two ways to provide an inspector:

1. Simple Inspection: Provide a function with the signature PeerScoreInspectFn, which receives a map[peer.ID]float64 containing the current score for each peer.
2. Extended Inspection: Provide a function with the signature ExtendedPeerScoreInspectFn, which receives a map[peer.ID]*PeerScoreSnapshot. This allows you to inspect individual components of the score, such as topic-specific stats, IP colocation factors, and behavioral penalties.

Note: WithPeerScoreInspect must be passed after the WithPeerScore option in your router configuration.

// Example using ExtendedPeerScoreInspectFn
inspector := func(scores map[peer.ID]*pubsub.PeerScoreSnapshot) {
    for id, snapshot := range scores {
        fmt.Printf("Peer %s score: %f (IP Colocation: %f)\n", id, snapshot.Score, snapshot.IPColocationFactor)
    }
}

// When initializing your PubSub/GossipSub router:
// ps, err := pubsub.NewGossipSub(host, 
//     pubsub.WithPeerScore(params), 
//     pubsub.WithPeerScoreInspect(inspector, 10*time.Second),
// )