Written by Joohhnnn

Libp2P applications in Optimism

In this section, the main purpose is to explain how Optimism uses Libb2P to complete the establishment of P2P networks in Op-Node. P2P networks are mainly used to pass information in different nodes, such as sequencers after completing the block construction of Unsafe, and propagating through the pub/sub of P2P's gossiphub. Libb2P also deals with other infrastructures such as networking, addressing, etc. in P2P networks.

Learn about libp2p

libp2p (abbreviated from "library peer-to-peer" or "library peer-to-peer") is a peer-to-peer (P2P) networking framework that helps develop P2P applications. It contains a set of protocols, specifications, and libraries that make P2P communication between network participants (also known as "peers" or "peers") easier (source[2] [3])。 Originally part of the IPFS (InterPlanetary File) project, libp2p evolved into a stand-alone project that became a modular network stack (source) for distributed networks )。

libp2p is an open source project of the IPFS community that welcomes contributions from a wide range of communities, including helping to write specifications, code implementations, and create examples and tutorials[4] [5])。 libp2p is made up of multiple building blocks, each with a very clear, documented, and tested interface that makes them composable, replaceable, and therefore upgradeable )。 The modular nature of libp2p allows developers to select and use only the components necessary for their applications, promoting flexibility and efficiency when building P2P web applications.

Related resources

• Official documentation for libp2p[6]
• libp2p's GitHub repository[7]
• Introduction to libp2p on ProtoSchool[8]

The modular architecture and open source nature of libp2p provide a good environment for developing powerful, scalable, and flexible P2P applications, making it an important player in the field of distributed networks and network application development.

libp2p implementation

When using libp2p, you will need to implement and configure some core components to build your P2P network. Here are some of the main implementation aspects of libp2p in the application:

1. Node Creation and Configuration:

• Creating and configuring a libp2p node is the most basic step, which includes setting the node's network address, identity, and other basic parameters. Key usage codes:

libp2p. New()

2. Transport Protocol:

Choose and configure your transport protocols (e.g. TCP, WebSockets, etc.) to ensure communication between nodes. Key usage codes:

tcpTransport := tcp. NewTCPTransport()

3. Multiplexing and Flow Control:

• Implement multiplexing to allow multiple concurrent data streams to be processed on a single connection.
• Implement flow control to manage the data transmission rate and processing rate. Key usage codes:

yamuxTransport := yamux. New()

4. Security and encryption:

• Configure a secure transport layer to ensure the security and privacy of communications.
• Implement encryption and authentication mechanisms to protect data and authenticate communicating parties. Key usage codes:

tlsTransport := tls. New()

5. Protocols and Message Handling:

Define and implement custom protocols to handle specific network operations and message exchanges.

• Process received messages and send responses as needed. Key usage codes:

host. SetStreamHandler("/my-protocol/1.0.0", myProtocolHandler)

6. Discovery and Routing:

• Implement a node discovery mechanism to find other nodes in the network.
• Implement routing logic to determine how messages are routed to the correct nodes in the network. Key usage codes:

dht := kaddht. NewDHT(ctx, host, datastore. NewMapDatastore())

7. Network behavior and policies:

Define and implement network behaviors and policies, such as connection management, error handling, and retry logic. Key usage codes:

connManager := connmgr. NewConnManager(lowWater, highWater, gracePeriod)

8. State Management and Storage:

Manage the status of nodes and networks, including connection status, node list, and data storage. Key usage codes:

peerstore := pstoremem. NewPeerstore()

9. Test and debug:

• Write tests for your libp2p application to ensure its correctness and reliability.
• Use debugging tools and logs to diagnose and resolve network issues. Key usage codes:

logging. SetLogLevel("libp2p", "DEBUG")

10. Documentation and Community Support:

• Consult libp2p's documentation to learn about its various components and APIs.
• Communicate with the libp2p community for support and problem solving.

}

The above are some of the main aspects to consider and implement when using libp2p. The specific implementation of each project may vary, but these basic aspects are necessary to build and run libp2p applications. When implementing these functions, you can refer to the official documentation of libp2p[9] [10]and GitHub repositories Sample code and tutorials.

Use of libp2p in OP-node

In order to figure out the relationship between op-node and libp2p, we have to figure out several questions

• Why libp2p? Why not choose devp2p (geth uses devp2p)
• What data or processes are closely related to the P2P network in OP-node
• How these features are implemented at the code layer

Why op-nodes need a libp2p network

First we need to understand why Optimism requires a P2P network ** LibbP2P is a modular network protocol that allows developers to build decentralized peer-to-peer applications for multiple use cases (sources[11] [12])(source [13])。 DevP2P, on the other hand, is mainly used in the Ethereum ecosystem and is tailored for Ethereum applications (Source )。 The flexibility and wide applicability of libp2p may make it a top choice for developers.

Op-Node mainly uses libp2p function points

• Used by the sequencer to pass the resulting unsafe block to other non-sequencer nodes
• For other nodes in non-sequencer mode for fast synchronization when a gap occurs (reverse chain synchronization)
• A good environment for adopting an integral reputation system to regulate the overall node

Code implementation

host custom initialization

Host can be understood as a P2P node, when opening this node, you need to make some special initialization configuration for your own project

Now let's look at the Host method in the op-node/p2p/host.go file.

This function is mainly used to set up the libp2p host and make various configurations. Here are the key parts of the function and a simple Chinese description of each:

1. Check if P2P is disabled
   If P2P is disabled, the function returns directly.
2. Get Peer ID from Public Key
   Use the public key in the configuration to generate the Peer ID.
3. Initialize Peerstore
   Create a base Peerstore store.
4. Initialize the extension Peerstore
   On top of the base Peerstore, create an extended Peerstore.
5. Add private and public keys to Peerstore
   Store the peer's private and public keys in the peerstore.
6. Initialize Connection Gater
   Used to control network connections.
7. Initialize Connection Manager
   Used to manage network connections.
8. Set Transmission and Listening Address
   Set the network transport protocol and the listening address of the host.
9. Create libp2p host
   Use all the previous settings to create a new libp2p host.
10. Initialize static peer
    If you have a static peer configured, initialize it.
11. Back to Host
    Finally, the function returns the created libp2p host.

These key sections are responsible for initializing and setting up the libp2p host, and each part is responsible for a specific aspect of the host configuration.

func (conf *Config) Host(log log. Logger, reporter metrics. Reporter, metrics HostMetrics) (host. Host, error) {  
    if conf. DisableP2P {  
        return nil, nil  
    }  
    pub := conf. Priv.GetPublic()  
    pid, err := peer. IDFromPublicKey(pub)  
    if err != nil {  
        return nil, fmt. Errorf("failed to derive pubkey from network priv key: %w", err)  
    }  
    basePs, err := pstoreds. NewPeerstore(context. Background(), conf. Store, pstoreds. DefaultOpts())  
    if err != nil {  
        return nil, fmt. Errorf("failed to open peerstore: %w", err)  
    }  
    peerScoreParams := conf. PeerScoringParams()  
     scoreRetention time. Duration  
    if peerScoreParams != nil {  
        // Use the same retention period as gossip will if available  
        scoreRetention = peerScoreParams.PeerScoring.RetainScore  
    } else {  
        // Disable score GC if peer scoring is disabled  
        scoreRetention = 0  
    }  
    ps, err := store. NewExtendedPeerstore(context. Background(), log, clock. Clock, basePs, conf. Store, scoreRetention)  
    if err != nil {  
        return nil, fmt. Errorf("failed to open extended peerstore: %w", err)  
    }  
    if err := ps. AddPrivKey(pid, conf. Priv); err != nil {  
        return nil, fmt. Errorf("failed to set up peerstore with priv key: %w", err)  
    }  
    if err := ps. AddPubKey(pid, pub); err != nil {  
        return nil, fmt. Errorf("failed to set up peerstore with pub key: %w", err)  
    }  
     connGtr gating. BlockingConnectionGater  
    connGtr, err = gating. NewBlockingConnectionGater(conf. Store)  
    if err != nil {  
        return nil, fmt. Errorf("failed to open connection gater: %w", err)  
    }  
    connGtr = gating. AddBanExpiry(connGtr, ps, log, clock. Clock, metrics)  
    connGtr = gating. AddMetering(connGtr, metrics)  
    connMngr, err := DefaultConnManager(conf)  
    if err != nil {  
        return nil, fmt. Errorf("failed to open connection manager: %w", err)  
    }  
    listenAddr, err := addrFromIPAndPort(conf. ListenIP, conf. ListenTCPPort)  
    if err != nil {  
        return nil, fmt. Errorf("failed to make listen addr: %w", err)  
    }  
    tcpTransport := libp2p. Transport(  
        tcp. NewTCPTransport,  
        tcp. WithConnectionTimeout(time. Minute*60)) // break unused connections  
    // TODO: technically we can also run the node on websocket and QUIC transports Maybe in the future?  
     nat lconf. NATManagerC // disabled if nil  
    if conf. NAT {  
        nat = basichost. NewNATManager  
    }  
    opts := []libp2p. Option{  
        libp2p. Identity(conf. Priv),  
        // Explicitly set the user-agent, so we can differentiate from other Go libp2p users.  
        libp2p. UserAgent(conf. UserAgent),  
        tcpTransport,  
        libp2p. WithDialTimeout(conf. TimeoutDial),  
        // No relay services, direct connections between peers only.  
        libp2p. DisableRelay(),  
        // host will start and listen to network directly after construction from config.  
        libp2p. ListenAddrs(listenAddr),  
        libp2p. ConnectionGater(connGtr),  
        libp2p. ConnectionManager(connMngr),  
        //libp2p. ResourceManager(nil), // TODO use resource manager interface to manage resources per peer better.  
        libp2p. Ping(true),  
        libp2p. AutoNATServiceRateLimit(10, 5, time. Second*60),  
    }  
    if conf. NoTransportSecurity {  
    } else {  
        opts = append(opts, conf. HostSecurity...)  
    }  
    h, err := libp2p. New(opts...)  
    if err != nil {  
    }  
    for i, peerAddr := range conf. StaticPeers {  
        if err != nil {  
        }  
        staticPeers[i]  = addr  
    }  
    out := &extraHost{  
        Host:        h,  
        staticPeers: staticPeers,  
    }  
        go out.monitorStaticPeers()  
    }  






        …  
        n.gs, err = NewGossipSub(resourcesCtx, n.host, rollupCfg, setup, n.scorer, metrics, log)  
        if err != nil {  
        }  
        …  

1. Authenticator Creation: Block Theme Name Generation:

• Generate blocksTopicName using the blocksTopicV1 function, which formats a string based on the L2ChainID in the configuration (cfg). Formatted strings follow a specific structure: /optimism/{L2ChainID}/0/blocks.

• function by calling ps. Join(blocksTopicName) attempts to add a block gossip theme. If an error occurs, it returns an error message indicating that the topic could not be joined.

• Generated a new goroutine to log topic events using the LogTopicEvents function.

• Created a subscriber using the MakeSubscriber function, which encapsulates a BlocksHandler that handles OnUnsafeL2Payload events from gossipIn. A new goroutine was generated to run the provided subion.

  func JoinGossip(p2pCtx context. Context, self peer.ID, ps *pubsub. PubSub, log log. Logger, cfg *rollup. Config, runCfg GossipRuntimeConfig, gossipIn GossipIn) (GossipOut, error) {
  blocksTopicName := blocksTopicV1(cfg) // return fmt. Sprintf("/optimism/%s/0/blocks", cfg. L2ChainID.String())
  return nil, fmt. Errorf("failed to register blocks gossip topic: %w", err)
  }
  blocksTopic, err := ps. Join(blocksTopicName)
  if err != nil {
  }
  if err != nil {
  }
  go LogTopicEvents(p2pCtx, log. New("topic", "blocks"), blocksTopicEvents)
  if err != nil {
  }
  return &publisher{log: log, cfg: cfg, blocksTopic: blocksTopic, runCfg: runCfg}, nil

op-node/rollup/driver/state.go

func (s *Driver) eventLoop() {  
    for(){  
        …  
            payload, err := s.sequencer.RunNextSequencerAction(ctx)  
            if err != nil {  
                // Errors are not severe enough to change/halt sequencing but should be logged and metered.  
                if err := s.network.PublishL2Payload(ctx, payload); err != nil {  
            }  
            planSequencerAction() // schedule the next sequencer action to keep the sequencing looping  
            }  
    }  
    …  

When there are missing blocks, fast synchronization via P2P

When nodes are relinked after downtime due to special circumstances, such as relinking after downtime, some blocks (gaps) may not be synchronized, and when encountering this situation, they can quickly synchronize through the reverse chain of the P2P network.

Let's take a look at the checkForGapInUnsafeQueue function in op-node/rollup/driver/state.go

The code snippet defines a method called checkForGapInUnsafeQueue, which belongs to the Driver structure. Its purpose is to check for data gaps in a queue called "unsafe queue" and attempt to retrieve missing payloads through an alternate synchronization method called altSync. The key point here is that the method is to ensure data continuity and try to retrieve missing data from other synchronization methods when missing data is detected. Here are the main steps of the function:

1. The function first gets UnsafeL2Head and UnsafeL2SyncTarget from s.derivation as the start and end points of the check range.

2. The function checks for missing blocks between start and end, which is done by comparing the Number values of end and start.

3. If a data gap is detected, the function will request the missing data range by calling s.altSync.RequestL2Range(ctx, start, end). If end is a null reference (i.e. eth. L2BlockRef{}), the function will request an open-end range synchronization, starting with start.

4. When requesting data, the function logs a debug log indicating which range of data it is requesting.

5. The function eventually returns an error value. If there are no errors, it returns nil

   // checkForGapInUnsafeQueue checks if there is a gap in the unsafe queue and attempts to retrieve the missing payloads from an alt-sync method.
   // WARNING: This is only an outgoing signal, the blocks are not guaranteed to be retrieved.
   // Results are received through OnUnsafeL2Payload.
   func (s *Driver) checkForGapInUnsafeQueue(ctx context. Context) error {
   start := s.derivation.UnsafeL2Head()
   end := s.derivation.UnsafeL2SyncTarget()
   // Check if we have missing blocks between the start and end Request them if we do.
   if end == (eth. L2BlockRef{}) {
   s.log.Debug("requesting sync with open-end range", "start", start)
   return s.altSync.RequestL2Range(ctx, start, eth. L2BlockRef{})
   } else if end. Number > start. Number+1 {
   s.log.Debug("requesting missing unsafe L2 block range", "start", start, "end", end, "size", end. Number-start.Number)
   return s.altSync.RequestL2Range(ctx, start, end)
   }
   return nil
   }

The RequestL2Range function signals the start and end of the request block to the requests channel.

The request is then distributed to the peerRequests channel via the onRangeRequest method, and the peerRequests channel is waited for by the loop opened by multiple peers, i.e. only one peer will process the request for each distribution.

func (s *SyncClient) onRangeRequest(ctx context. Context, req rangeRequest) {  
        …  
        for i := uint64(0); ; i++ {  
        num := req.end.Number - 1 - i  
        if num <= req.start {  
            return  
        }  
        // check if we have something in quarantine already  
        if h, ok := s.quarantineByNum[num] ; ok {  
            if s.trusted.Contains(h) { // if we trust it, try to promote it.  
                s.tryPromote(h)  
            }  
            // Don't fetch things that we have a candidate for already.  
            // We'll evict it from quarantine by finding a conflict, or if we sync enough other blocks  
            continue  
        }  
        if _, ok := s.inFlight[num] ; ok {  
            log. Debug("request still in-flight, not rescheduling sync request", "num", num)  
            continue // request still in flight  
        }  
        pr := peerRequest{num: num, complete: new(atomic. Bool)}  
        log. Debug("Scheduling P2P block request", "num", num)  
        // schedule number  
        select {  
        case s.peerRequests <- pr:  
            s.inFlight[num]  = pr.complete  
        case <-ctx. Done():  
            log. Info("did not schedule full P2P sync range", "current", num, "err", ctx. Err())  
            return  
        default: // peers may all be busy processing requests already  
            log. Info("no peers ready to handle block requests for more P2P requests for L2 block history", "current", num)  
            return  
        }  
    }  
}

Let's see what happens when the peer receives this request.

The first thing we need to know is that the link between the peer and the requesting node, or the message passing, is passed through the libp2p stream. The processing method of the stream is implemented by the receiving peer node, and the creation of the stream is opened by the sending node.

We can see this code in the previous init function, where MakeStreamHandler returns a handler, and SetStreamHandler binds the protocol ID to this handler, so whenever the sending node creates and uses this stream, the returned handler is triggered.

n.syncSrv = NewReqRespServer(rollupCfg, l2Chain, metrics)  
// register the sync protocol with libp2p host  
payloadByNumber := MakeStreamHandler(resourcesCtx, log. New("serve", "payloads_by_number"), n.syncSrv.HandleSyncRequest)  
n.host.SetStreamHandler(PayloadByNumberProtocolID(rollupCfg.L2ChainID), payloadByNumber)

Let's look at how it is handled inside the handler The function first performs global and individual rate limit checks to control how quickly requests are processed. It then reads and verifies the requested block number, ensuring that it is within a reasonable range. The function then gets the chunk payload of the request from the L2 layer and writes it to the response stream. When writing response data, it sets a write deadline to avoid being blocked by slow peer connections during the write process. Finally, the function returns the requested block number and possible errors.

func (srv *ReqRespServer) handleSyncRequest(ctx context. Context, stream network. Stream) (uint64, error) {  
    peerId := stream. Conn(). RemotePeer()  
    // take a token from the global rate-limiter,  
    // to make sure there's not too much concurrent server work between different peers.  
    if err := srv.globalRequestsRL.Wait(ctx); err != nil {  
        return 0, fmt. Errorf("timed out waiting for global sync rate limit: %w", err)  
    }  
    // find rate limiting data of peer, or add otherwise  
    srv.peerStatsLock.Lock()  
    ps, _ := srv.peerRateLimits.Get(peerId)  
    if ps == nil {  
        ps = &peerStat{  
            Requests: rate. NewLimiter(peerServerBlocksRateLimit, peerServerBlocksBurst),  
        }  
        srv.peerRateLimits.Add(peerId, ps)  
        ps. Requests.Reserve() // count the hit, but make it delay the next request rather than immediately waiting  
    } else {  
        // Only wait if it's an existing peer, otherwise the instant rate-limit Wait call always errors.  
        // If the requester thinks we're taking too long, then it's their problem and they can disconnect.  
        // We'll disconnect ourselves only when failing to read/write,  
        // if the work is invalid (range validation), or when individual sub tasks timeout.  
        if err := ps. Requests.Wait(ctx); err != nil {  
            return 0, fmt. Errorf("timed out waiting for global sync rate limit: %w", err)  
        }  
    }  
    srv.peerStatsLock.Unlock()  
    // Set read deadline, if available  
    _ = stream. SetReadDeadline(time. Now(). Add(serverReadRequestTimeout))  
    // Read the request  
     req uint64  
    if err := binary. Read(stream, binary. LittleEndian, &req); err != nil {  
        return 0, fmt. Errorf("failed to read requested block number: %w", err)  
    }  
    if err := stream. CloseRead(); err != nil {  
        return req, fmt. Errorf("failed to close reading-side of a P2P sync request call: %w", err)  
    }  
    // Check the request is within the expected range of blocks  
    if req < srv.cfg.Genesis.L2.Number {  
        return req, fmt. Errorf("cannot serve request for L2 block %d before genesis %d: %w", req, srv.cfg.Genesis.L2.Number, invalidRequestErr)  
    }  
    max, err := srv.cfg.TargetBlockNumber(uint64(time. Now(). Unix()))  
    if err != nil {  
        return req, fmt. Errorf("cannot determine max target block number to verify request: %w", invalidRequestErr)  
    }  
    if req > max {  
        return req, fmt. Errorf("cannot serve request for L2 block %d after max expected block (%v): %w", req, max, invalidRequestErr)  
    }  
    payload, err := srv.l2.PayloadByNumber(ctx, req)  
    if err != nil {  
        if errors. Is(err, ethereum. NotFound) {  
            return req, fmt. Errorf("peer requested unknown block by number: %w", err)  
        } else {  
            return req, fmt. Errorf("failed to retrieve payload to serve to peer: %w", err)  
        }  
    }  
    // We set write deadline, if available, to safely write without blocking on a throttling peer connection  
    _ = stream. SetWriteDeadline(time. Now(). Add(serverWriteChunkTimeout))  
    // 0 - resultCode: success = 0  
    // 1:5 - version: 0  
     tmp [5] byte  
    if _, err := stream. Write(tmp[:]); err != nil {  
        return req, fmt. Errorf("failed to write response header data: %w", err)  
    }  
    w := snappy. NewBufferedWriter(stream)  
    if _, err := payload. MarshalSSZ(w); err != nil {  
        return req, fmt. Errorf("failed to write payload to sync response: %w", err)  
    }  
    if err := w.Close(); err != nil {  
        return req, fmt. Errorf("failed to finishing writing payload to sync response: %w", err)  
    }  
    return req, nil  
}

At this point, the general process of reverse chain synchronization request and processing has been explained

Points reputation system in p2p nodes

In order to prevent certain nodes from making malicious requests and responses that undermine the security of the entire network, Optimism also uses a points system.

For example, in op-node/p2p/app_scores.go, there are a series of functions to set the score of a peer

func (s *peerApplicationScorer) onValidResponse(id peer.ID) {  
    _, err := s.scorebook.SetScore(id, store. IncrementValidResponses{Cap: s.params.ValidResponseCap})  
    if err != nil {  
        s.log.Error("Unable to update peer score", "peer", id, "err", err)  
        return  
    }  
}  
func (s *peerApplicationScorer) onResponseError(id peer.ID) {  
    _, err := s.scorebook.SetScore(id, store. IncrementErrorResponses{Cap: s.params.ErrorResponseCap})  
    if err != nil {  
        s.log.Error("Unable to update peer score", "peer", id, "err", err)  
        return  
    }  
}  
func (s *peerApplicationScorer) onRejectedPayload(id peer.ID) {  
    _, err := s.scorebook.SetScore(id, store. IncrementRejectedPayloads{Cap: s.params.RejectedPayloadCap})  
    if err != nil {  
        s.log.Error("Unable to update peer score", "peer", id, "err", err)  
        return  
    }  
}

The integration status of the new node is then checked before it is added

func AddScoring(gater BlockingConnectionGater, scores Scores, minScore float64) *ScoringConnectionGater {  
    return &ScoringConnectionGater{BlockingConnectionGater: gater, scores: scores, minScore: minScore}  
}  
func (g *ScoringConnectionGater) checkScore(p peer.ID) (allow bool) {  
    score, err := g.scores.GetPeerScore(p)  
    if err != nil {  
        return false  
    }  
    return score >= g.minScore  
}  
func (g *ScoringConnectionGater) InterceptPeerDial(p peer.ID) (allow bool) {  
    return g.BlockingConnectionGater.InterceptPeerDial(p) && g.checkScore(p)  
}  
func (g *ScoringConnectionGater) InterceptAddrDial(id peer.ID, ma multiaddr. Multiaddr) (allow bool) {  
    return g.BlockingConnectionGater.InterceptAddrDial(id, ma) && g.checkScore(id)  
}  
func (g *ScoringConnectionGater) InterceptSecured(dir network. Direction, id peer.ID, mas network. ConnMultiaddrs) (allow bool) {  
    return g.BlockingConnectionGater.InterceptSecured(dir, id, mas) && g.checkScore(id)  
}

Summary

The high configurability of libp2p makes the entire project p2p highly customizable and modular, the above is the main logic of optimsim's personalized implementation of libp2p, and other details can be learned in detail by reading the source code in the p2p directory.