Go P2P Development: From Basics to PubSub

June 5, 2023 Network P2P, Golang, Libp2p, DHT, PubSub, Custom Protocol Network Development 1931 words 10 min read

Go’s simple concurrency model and fast compilation make it a popular choice for P2P network development. go-libp2p is one of the most complete libp2p implementations and is widely used in large projects like IPFS (Kubo).

Environment Setup

bash
1
2
3
go get github.com/libp2p/go-libp2p
go get github.com/libp2p/go-libp2p-kad-dht
go get github.com/libp2p/go-libp2p-pubsub

go.mod configuration:

1
2
3
4
5
6
7
8
9
module p2p-tutorial

go 1.21

require (
    github.com/libp2p/go-libp2p v0.35.0
    github.com/libp2p/go-libp2p-kad-dht v0.25.0
    github.com/libp2p/go-libp2p-pubsub v0.10.0
)

Building Host and Node Interconnection

Go’s libp2p host creation uses the options pattern. Here’s a complete Echo node with host creation, stream handler, and node interconnection:

go
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
package main

import (
    "context"
    "fmt"
    "io"
    "os"
    "os/signal"
    "syscall"

    "github.com/libp2p/go-libp2p"
    "github.com/libp2p/go-libp2p/core/host"
    "github.com/libp2p/go-libp2p/core/network"
    "github.com/libp2p/go-libp2p/core/peer"
)

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    h, err := libp2p.New(
        libp2p.ListenAddrStrings("/ip4/0.0.0.0/tcp/0"),
        libp2p.NATPortMap(),
    )
    if err != nil {
        panic(err)
    }
    defer h.Close()

    fmt.Printf("Host ID: %s\n", h.ID())
    fmt.Println("Listening addresses:")
    for _, addr := range h.Addrs() {
        fmt.Printf("  %s/p2p/%s\n", addr, h.ID())
    }

    // Set stream handler (Echo service)
    h.SetStreamHandler("/echo/1.0.0", func(s network.Stream) {
        fmt.Printf("New stream from %s\n", s.Conn().RemotePeer())
        go func() {
            io.Copy(s, s)  // Echo: return received data as-is
            s.Close()
        }()
    })

    // If a peer address is provided as CLI arg, connect to it
    if len(os.Args) > 1 {
        peerAddr := os.Args[1]
        fmt.Printf("Dialing %s\n", peerAddr)

        peerInfo, err := peer.AddrInfoFromString(peerAddr)
        if err != nil {
            panic(err)
        }

        // Connect to peer
        if err := h.Connect(ctx, *peerInfo); err != nil {
            panic(err)
        }
        fmt.Printf("Connected to %s\n", peerInfo.ID)

        // Open a stream and test Echo
        s, err := h.NewStream(ctx, peerInfo.ID, "/echo/1.0.0")
        if err != nil {
            panic(err)
        }
        s.Write([]byte("Hello from Go!\n"))

        buf := make([]byte, 1024)
        n, _ := s.Read(buf)
        fmt.Printf("Echo response: %s", string(buf[:n]))
        s.Close()
    }

    // Wait for interrupt signal
    ch := make(chan os.Signal, 1)
    signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM)
    <-ch
    fmt.Println("Shutting down...")
}

Running and testing:

bash
1
2
3
4
5
6
7
8
9
# Terminal 1: Start first node
go run main.go
# Output: Host ID: 12D3Koo...
# Output: Listening addresses: /ip4/127.0.0.1/tcp/12345/p2p/12D3Koo...

# Terminal 2: Start second node and connect to first
go run main.go /ip4/127.0.0.1/tcp/12345/p2p/12D3Koo...
# Output: Connected to 12D3Koo...
# Output: Echo response: Hello from Go!

Context Management

Go’s context.Context is the core mechanism for lifecycle control in P2P development:

go
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
// Create cancellable context (for entire app lifecycle)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()

// Create timeout context (for single operations)
dialCtx, dialCancel := context.WithTimeout(ctx, 10*time.Second)
defer dialCancel()

// All operations using this context abort on timeout/cancel
h.Connect(dialCtx, peerInfo)

All go-libp2p network operations (Connect, NewStream, PutValue, etc.) accept context.Context as the first parameter. Proper context usage prevents goroutine leaks — when the node needs to shut down gracefully, canceling the top-level context aborts all in-flight operations.

Kademlia DHT Implementation

go
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
package main

import (
    "context"
    "fmt"
    "time"

    "github.com/libp2p/go-libp2p"
    "github.com/libp2p/go-libp2p/core/host"
    dht "github.com/libp2p/go-libp2p-kad-dht"
)

func NewDHTHost(ctx context.Context) (host.Host, *dht.IpfsDHT, error) {
    h, err := libp2p.New(
        libp2p.ListenAddrStrings("/ip4/0.0.0.0/tcp/4001"),
    )
    if err != nil {
        return nil, nil, err
    }

    // Create DHT in server mode (participates in routing)
    kadDHT, err := dht.New(ctx, h,
        dht.Mode(dht.ModeServer),
        dht.BootstrapPeers(dht.GetDefaultBootstrapPeerAddrInfos()...),
    )

    if err = kadDHT.Bootstrap(ctx); err != nil {
        return nil, nil, err
    }
    return h, kadDHT, nil
}

func dhtOperations(ctx context.Context, kadDHT *dht.IpfsDHT) error {
    // ===== Store key-value pair =====
    key := "/p2p-tutorial/example-key"
    err := kadDHT.PutValue(ctx, key, []byte("Hello DHT!"),
        dht.Quorum(2),
    )
    if err != nil {
        return err
    }
    fmt.Println("Put: stored value for key")

    // ===== Retrieve value =====
    retrieved, err := kadDHT.GetValue(ctx, key,
        dht.Quorum(1),
        dht.MaxRecordAge(1*time.Hour),
    )
    if err != nil {
        return err
    }
    fmt.Printf("Get: retrieved value = %s\n", string(retrieved))

    // ===== Find content providers =====
    contentID := "QmYyQSo1c1Ym7orWxLYvCrM2EmxFTANf8wXmmE7DWjhx5N"
    providers := kadDHT.FindProvidersAsync(ctx, contentID, 10)
    for pi := range providers {
        fmt.Printf("Provider found: %s\n", pi.ID)
    }
    return nil
}

PubSub Publish/Subscribe

go
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
package main

import (
    "context"
    "fmt"
    "time"

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

func runPubSub(ctx context.Context) error {
    h, _ := libp2p.New(libp2p.ListenAddrStrings("/ip4/0.0.0.0/tcp/0"))
    defer h.Close()

    // Create GossipSub instance
    ps, err := pubsub.NewGossipSub(ctx, h,
        pubsub.WithMessageSignaturePolicy(pubsub.StrictSign),
    )
    if err != nil {
        return err
    }

    // Join topic
    topic, err := ps.Join("p2p-tutorial/chat")
    if err != nil {
        return err
    }

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

    // Background publisher
    go func() {
        for i := 0; i < 5; i++ {
            msg := fmt.Sprintf("Message %d from %s", i, h.ID())
            topic.Publish(ctx, []byte(msg))
            time.Sleep(1 * time.Second)
        }
    }()

    // Receive loop
    for {
        msg, err := sub.Next(ctx)
        if err != nil {
            return err  // Exits when context is cancelled
        }
        // Skip own messages
        if msg.ReceivedFrom == h.ID() {
            continue
        }
        fmt.Printf("[%s]: %s\n", msg.GetFrom(), string(msg.Data))
    }
}

Note: sub.Next(ctx) blocks until a message is received or context is cancelled. When ctx is cancelled (e.g., app shutdown), Next returns an error and the receive loop exits automatically — this is key to graceful shutdown.

Here’s a complete custom protocol with both request and response sides:

go
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
package main

import (
    "context"
    "encoding/binary"
    "fmt"
    "io"
    "os"

    "github.com/libp2p/go-libp2p/core/host"
    "github.com/libp2p/go-libp2p/core/network"
    "github.com/libp2p/go-libp2p/core/peer"
    "github.com/libp2p/go-libp2p/core/protocol"
)

const FileShareProtocolID = protocol.ID("/p2p-tutorial/file-share/1.0.0")

type FileShareProtocol struct {
    host host.Host
}

func NewFileShareProtocol(h host.Host) *FileShareProtocol {
    fs := &FileShareProtocol{host: h}
    // Register stream handler (server-side logic)
    h.SetStreamHandler(FileShareProtocolID, fs.handleStream)
    return fs
}

// handleStream handles inbound connections (server: responds to file requests)
func (fs *FileShareProtocol) handleStream(s network.Stream) {
    defer s.Close()

    var msgType uint8
    if err := binary.Read(s, binary.BigEndian, &msgType); err != nil {
        return
    }

    if msgType == 1 { // File request
        var nameLen uint32
        binary.Read(s, binary.BigEndian, &nameLen)
        nameBuf := make([]byte, nameLen)
        io.ReadFull(s, nameBuf)
        filename := string(nameBuf)

        fmt.Printf("File requested: %s\n", filename)

        data, err := os.ReadFile(filename)
        if err != nil {
            binary.Write(s, binary.BigEndian, uint8(3)) // Error response
            return
        }
        binary.Write(s, binary.BigEndian, uint8(2)) // Success response
        binary.Write(s, binary.BigEndian, uint32(len(data)))
        s.Write(data)
    }
}

// RequestFile requests a file (client-side logic)
func (fs *FileShareProtocol) RequestFile(
    ctx context.Context,
    peerID peer.ID,
    filename string,
) ([]byte, error) {
    s, err := fs.host.NewStream(ctx, peerID, FileShareProtocolID)
    if err != nil {
        return nil, err
    }
    defer s.Close()

    // Send request
    binary.Write(s, binary.BigEndian, uint8(1))              // Request type
    binary.Write(s, binary.BigEndian, uint32(len(filename))) // Filename length
    s.Write([]byte(filename))                                 // Filename

    // Read response
    var respType uint8
    binary.Read(s, binary.BigEndian, &respType)
    if respType == 3 {
        return nil, fmt.Errorf("peer could not read file")
    }
    var dataLen uint32
    binary.Read(s, binary.BigEndian, &dataLen)
    data := make([]byte, dataLen)
    io.ReadFull(s, data)
    return data, nil
}

Architecture Comparison

mermaid
flowchart LR
    subgraph Rust
        RR["Rust libp2p<br/>Derive macro composition<br/>tokio event loop<br/>Compile-time safety"]
    end
    subgraph Go
        GG["Go libp2p<br/>Options pattern config<br/>goroutine concurrency<br/>Runtime safety"]
    end

    RR -->|"NetworkBehaviour Trait"| RBeh["Protocol composition<br/>Type safe"]
    GG -->|"Option Pattern"| GBeh["Flexible config<br/>Fast iteration"]

    style RR fill:#FFD43B
    style GG fill:#00ADD8,color:#fff

Production Case Study: Syncthing’s Go Architecture

Syncthing is the most influential open-source project in Go-based P2P file synchronization (85K+ stars, MPL-2.0), with ~200,000 lines of Go code. Unlike libp2p’s framework approach, Syncthing is a complete end-to-end application whose architecture offers an alternative paradigm for building P2P systems in Go.

BEP Protocol vs libp2p Streams

Syncthing uses its own BEP (Block Exchange Protocol) over TLS 1.3 with Protobuf serialization:

protobuf
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
message FileInfo {
    string name         = 1;
    int64  size         = 3;
    int64  modified_s   = 5;
    Vector version      = 9;
    int64  sequence     = 10;
    repeated BlockInfo blocks = 16;
}

message BlockInfo {
    bytes hash   = 3;
    int64 offset = 1;
    int32 size   = 2;
}

Compared to libp2p’s multiplexed streams (multistream-select), BEP is closer to traditional TCP protocol design — each connection uses independent reader/writer goroutines with channel-based message coordination:

Dimension	Syncthing BEP	libp2p
Transport	TLS 1.3 encrypted TCP	Noise / TLS 1.3 + stream multiplexing
Serialization	Protobuf (strict schema)	Custom varint length prefix
Concurrency	goroutine-per-connection + channel	Stream multiplexing + ResourceManager
Identity	Certificate fingerprint SHA-256 (Device ID)	Peer ID (public key hash)
Protocol Discovery	ClusterConfig negotiation at handshake	multistream-select negotiation

Go Concurrency: Supervision Tree

Syncthing uses the suture library for a Supervision Tree architecture to manage component lifecycles — inspired by Erlang OTP:

mermaid
flowchart TD
    SM["suture.Supervisor"] --> FOLDERS["Folders"]
    SM --> NET["Network<br/>Services"]
    SM --> DB["SQLite DB"]

    style SM fill:#2196F3,color:#fff
    style FOLDERS fill:#4CAF50,color:#fff
    style NET fill:#FF9800,color:#fff
    style DB fill:#f44336,color:#fff

Each folder component starts an independent goroutine loop (serve loop) receiving scan, pull, and retry events via channels. If a folder’s sync logic panics, the Supervisor automatically restarts it without affecting other folders.

In contrast, libp2p uses centralized resource management via ResourceManager — setting global caps on connections, memory, and file descriptors, rejecting new connections when exceeded. The former emphasizes fault recovery, while the latter focuses on resource isolation.

Buffer Pool and Memory Management

Syncthing makes extensive use of sync.Pool to reduce GC pressure:

go
1
2
3
4
5
6
7
var bufPool = sync.Pool{
    New: func() interface{} { return make([]byte, 32<<10) }, // 32KB buffer
}

var hashPool = sync.Pool{
    New: func() interface{} { return sha256.New() },
}

During file scanning, 32KB read/write buffers and SHA-256 hash instances are obtained from the Pool and returned after use. In large file sync scenarios, this significantly reduces heap allocation count and GC pause time.

This pattern complements libp2p’s ResourceManager — sync.Pool focuses on object reuse to reduce GC pressure, while ResourceManager enforces resource limits to prevent OOM. They can be combined: use Pool for high-frequency small objects, and ResourceManager for overall resource caps.

Multi-Layer Discovery System

Syncthing implements a three-tier progressive peer discovery mechanism:

mermaid
flowchart TD
    LAN["LAN Discovery<br/>UDP Broadcast"] -->|"Direct LAN"| CON["Connect"]
    GLOBAL["Global Discovery<br/>HTTP + HTTPS"] -->|"Cross-network"| CON
    RELAY["Relay Fallback<br/>TLS forwarding"] -->|"NAT failed"| CON

    style LAN fill:#4CAF50,color:#fff
    style GLOBAL fill:#2196F3,color:#fff
    style RELAY fill:#FF9800,color:#fff

The three layers probe in descending priority: first LAN broadcast (zero-config, instant discovery on same subnet), then global discovery servers (across the internet), and finally relay fallback when NAT hole punching fails. Relay servers never decrypt traffic — they only forward TLS-encrypted packets, consistent with libp2p’s Circuit Relay v2 design philosophy.

References

go-libp2p. https://github.com/libp2p/go-libp2p
go-libp2p-kad-dht. https://github.com/libp2p/go-libp2p-kad-dht
IPFS Kubo. https://github.com/ipfs/kubo
go-libp2p examples. https://github.com/libp2p/go-libp2p/tree/master/examples
Syncthing source. https://github.com/syncthing/syncthing
BEP Protocol Spec. https://docs.syncthing.net/specs/bep-v1.html
suture (Go supervision tree). https://github.com/thejerf/suture

Part of series: Network Development

← Previous Rust P2P Development: From Ping to Gossipsub Next → Hands-On: Building a Distributed File Sharing System