-
Notifications
You must be signed in to change notification settings - Fork 33
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[P2P] refactor: message handling (#763)
## Description
### Before
The P2P module is responsible for handling incoming messages from the
router's host. When it receives a message, it calls into the router to
unpack the `RainTreeMessage` (and facilitate further broadcast
propagation). The resulting `PocketEnvelope` is returned to the P2P
module to be emitted over the application event bus.
```mermaid
classDiagram
class RainTreeMessage {
<<protobuf>>
+Level uint32
+Data []byte
+Nonce uint64
}
class PocketEnvelope {
<<protobuf>>
+Content *anypb.Any
}
RainTreeMessage --* PocketEnvelope : serialized as `Data`
class P2PModule {
-handleNetworkData([]byte) error
-handleStream(stream libp2pNetwork.Stream)
-readStream(stream libp2pNetwork.Stream)
}
class RainTreeRouter {
+HandleNetworkData func([]byte) ([]byte, error)
}
RainTreeRouter --> P2PModule
P2PModule --> RainTreeRouter
RainTreeRouter --o RainTreeMessage
P2PModule --o PocketEnvelope
```
### After
The router encapsulates handling incoming `RainTreeMessage`s, unpacking
them and passing them along to the P2P module as serialized
`PocketEnvelope`s. The P2P module then deserializes and emits them over
the application event bus.
```mermaid
classDiagram
class RainTreeMessage {
<<protobuf>>
+Level uint32
+Data []byte
+Nonce uint64
}
class PocketEnvelope {
<<protobuf>>
+Content *anypb.Any
}
RainTreeMessage --* PocketEnvelope : serialized as `Data`
class P2PModule {
-handlePocketEnvelope([]byte) error
}
class RainTreeRouter {
-handler RouterHandler
-handleRainTreeMsg([]byte) ([]byte, error)
-handleStream(stream libp2pNetwork.Stream)
-readStream(stream libp2pNetwork.Stream)
}
RainTreeRouter --> P2PModule : `handler` == `handlePocketEnvelope`
RainTreeRouter --o RainTreeMessage
P2PModule --o PocketEnvelope
```
## Issue
Second deliverable in #762
## Type of change
Please mark the relevant option(s):
- [ ] New feature, functionality or library
- [ ] Bug fix
- [x] Code health or cleanup
- [ ] Major breaking change
- [ ] Documentation
- [ ] Other <!-- add details here if it a different type of change -->
## List of changes
- Added `Handler` field to `RainTreeConfig` & `BackgroundConfig`
- Refactored `rainTreeRouter` logging methods
- Renamed `rainTreeRouter#HandleNetworkData()` to `#handleRainTreeMsg()`
- Renamed `p2pModule#handleNetworkData()` to `#handleAppData()`
- Added -tags=test to all test make targets
- Fixed mockdns usage in `TestP2PModule_Insecure_Error` test
- Moved message handling from p2p module to router
## Testing
- [ ] `make develop_test`; if any code changes were made
- [ ] `make test_e2e` on [k8s
LocalNet](https://github.com/pokt-network/pocket/blob/main/build/localnet/README.md);
if any code changes were made
- [ ] `e2e-devnet-test` passes tests on
[DevNet](https://pocketnetwork.notion.site/How-to-DevNet-ff1598f27efe44c09f34e2aa0051f0dd);
if any code was changed
- [x] [Docker Compose
LocalNet](https://github.com/pokt-network/pocket/blob/main/docs/development/README.md);
if any major functionality was changed or introduced
- [x] [k8s
LocalNet](https://github.com/pokt-network/pocket/blob/main/build/localnet/README.md);
if any infrastructure or configuration changes were made
## Required Checklist
- [x] I have performed a self-review of my own code
- [x] I have commented my code, particularly in hard-to-understand areas
- [ ] I have added, or updated, [`godoc` format
comments](https://go.dev/blog/godoc) on touched members (see:
[tip.golang.org/doc/comment](https://tip.golang.org/doc/comment))
- [ ] I have tested my changes using the available tooling
- [ ] I have updated the corresponding CHANGELOG
### If Applicable Checklist
- [ ] I have updated the corresponding README(s); local and/or global
- [ ] I have added tests that prove my fix is effective or that my
feature works
- [ ] I have added, or updated,
[mermaid.js](https://mermaid-js.github.io) diagrams in the corresponding
README(s)
- [ ] I have added, or updated, documentation and
[mermaid.js](https://mermaid-js.github.io) diagrams in `shared/docs/*`
if I updated `shared/*`README(s)- Loading branch information
1 parent
5d1944c
commit fba11c3
Showing
11 changed files
with
210 additions
and
151 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -3,16 +3,11 @@ package p2p | |
| import ( | ||
| "errors" | ||
| "fmt" | ||
| "io" | ||
| "time" | ||
|
|
||
| "github.com/libp2p/go-libp2p" | ||
| libp2pHost "github.com/libp2p/go-libp2p/core/host" | ||
| libp2pNetwork "github.com/libp2p/go-libp2p/core/network" | ||
| "github.com/multiformats/go-multiaddr" | ||
| "github.com/pokt-network/pocket/logger" | ||
| "github.com/pokt-network/pocket/p2p/config" | ||
| "github.com/pokt-network/pocket/p2p/protocol" | ||
| "github.com/pokt-network/pocket/p2p/providers" | ||
| "github.com/pokt-network/pocket/p2p/providers/current_height_provider" | ||
| "github.com/pokt-network/pocket/p2p/providers/peerstore_provider" | ||
|
|
@@ -32,12 +27,6 @@ import ( | |
| "google.golang.org/protobuf/types/known/anypb" | ||
| ) | ||
|
|
||
| // TECHDEBT(#629): configure timeouts. Consider security exposure vs. real-world conditions. | ||
| // TECHDEBT(#629): parameterize and expose via config. | ||
| // readStreamTimeout is the duration to wait for a read operation on a | ||
| // stream to complete, after which the stream is closed ("timed out"). | ||
| const readStreamTimeout = time.Second * 10 | ||
|
|
||
| var _ modules.P2PModule = &p2pModule{} | ||
|
|
||
| type p2pModule struct { | ||
|
|
@@ -173,11 +162,6 @@ func (m *p2pModule) Start() (err error) { | |
| return fmt.Errorf("setting up router: %w", err) | ||
| } | ||
|
|
||
| // Don't handle incoming streams in client debug mode. | ||
| if !m.isClientDebugMode() { | ||
| m.host.SetStreamHandler(protocol.PoktProtocolID, m.handleStream) | ||
| } | ||
|
|
||
| m.GetBus(). | ||
| GetTelemetryModule(). | ||
| GetTimeSeriesAgent(). | ||
|
|
@@ -290,6 +274,7 @@ func (m *p2pModule) setupRouter() (err error) { | |
| CurrentHeightProvider: m.currentHeightProvider, | ||
| PeerstoreProvider: m.pstoreProvider, | ||
| Host: m.host, | ||
| Handler: m.handlePocketEnvelope, | ||
| MaxNonces: m.cfg.MaxNonces, | ||
| }, | ||
| ) | ||
|
|
@@ -340,100 +325,16 @@ func (m *p2pModule) isClientDebugMode() bool { | |
| return m.GetBus().GetRuntimeMgr().GetConfig().ClientDebugMode | ||
| } | ||
|
|
||
| // handleStream is called each time a peer establishes a new stream with this | ||
| // module's libp2p `host.Host`. | ||
| func (m *p2pModule) handleStream(stream libp2pNetwork.Stream) { | ||
| m.logger.Debug().Msg("handling incoming stream") | ||
| peer, err := utils.PeerFromLibp2pStream(stream) | ||
| if err != nil { | ||
| m.logger.Error().Err(err). | ||
| Str("address", peer.GetAddress().String()). | ||
| Msg("parsing remote peer identity") | ||
|
|
||
| if err = stream.Reset(); err != nil { | ||
| m.logger.Error().Err(err).Msg("resetting stream") | ||
| } | ||
| return | ||
| } | ||
|
|
||
| if err := m.router.AddPeer(peer); err != nil { | ||
| m.logger.Error().Err(err). | ||
| Str("address", peer.GetAddress().String()). | ||
| Msg("adding remote peer to router") | ||
| } | ||
|
|
||
| go m.readStream(stream) | ||
| } | ||
|
|
||
| // readStream is intended to be called in a goroutine. It continuously reads from | ||
| // the given stream for handling at the network level. Used for handling "direct" | ||
| // messages (i.e. one specific target node). | ||
| func (m *p2pModule) readStream(stream libp2pNetwork.Stream) { | ||
| // Time out if no data is sent to free resources. | ||
| if err := stream.SetReadDeadline(newReadStreamDeadline()); err != nil { | ||
| // NB: tests using libp2p's `mocknet` rely on this not returning an error. | ||
| // `SetReadDeadline` not supported by `mocknet` streams. | ||
| m.logger.Debug().Err(err).Msg("setting stream read deadline") | ||
| } | ||
|
|
||
| // debug logging: stream scope stats | ||
| // (see: https://pkg.go.dev/github.com/libp2p/[email protected]/core/network#StreamScope) | ||
| if err := utils.LogScopeStatFactory( | ||
| &logger.Global.Logger, | ||
| "stream scope (read-side)", | ||
| )(stream.Scope()); err != nil { | ||
| m.logger.Debug().Err(err).Msg("logging stream scope stats") | ||
| } | ||
| // --- | ||
|
|
||
| data, err := io.ReadAll(stream) | ||
| if err != nil { | ||
| m.logger.Error().Err(err).Msg("reading from stream") | ||
| if err := stream.Reset(); err != nil { | ||
| m.logger.Debug().Err(err).Msg("resetting stream (read-side)") | ||
| } | ||
| return | ||
| } | ||
|
|
||
| if err := stream.Reset(); err != nil { | ||
| m.logger.Debug().Err(err).Msg("resetting stream (read-side)") | ||
| } | ||
|
|
||
| // debug logging | ||
| remotePeer, err := utils.PeerFromLibp2pStream(stream) | ||
| if err != nil { | ||
| m.logger.Debug().Err(err).Msg("getting remote remotePeer") | ||
| } else { | ||
| utils.LogIncomingMsg(m.logger, m.cfg.Hostname, remotePeer) | ||
| } | ||
| // --- | ||
|
|
||
| if err := m.handleNetworkData(data); err != nil { | ||
| m.logger.Error().Err(err).Msg("handling network data") | ||
| } | ||
| } | ||
|
|
||
| // handleNetworkData passes a network message to the configured | ||
| // `Router`implementation for routing. | ||
| func (m *p2pModule) handleNetworkData(data []byte) error { | ||
| appMsgData, err := m.router.HandleNetworkData(data) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| // There was no error, but we don't need to forward this to the app-specific bus. | ||
| // For example, the message has already been handled by the application. | ||
| if appMsgData == nil { | ||
| return nil | ||
| } | ||
|
|
||
| networkMessage := messaging.PocketEnvelope{} | ||
| if err := proto.Unmarshal(appMsgData, &networkMessage); err != nil { | ||
| // handlePocketEnvelope deserializes the received `PocketEnvelope` data and publishes | ||
| // a copy of its `Content` to the application event bus. | ||
| func (m *p2pModule) handlePocketEnvelope(pocketEnvelopeBz []byte) error { | ||
| poktEnvelope := messaging.PocketEnvelope{} | ||
| if err := proto.Unmarshal(pocketEnvelopeBz, &poktEnvelope); err != nil { | ||
| return fmt.Errorf("decoding network message: %w", err) | ||
| } | ||
|
|
||
| event := messaging.PocketEnvelope{ | ||
| Content: networkMessage.Content, | ||
| Content: poktEnvelope.Content, | ||
| } | ||
| m.GetBus().PublishEventToBus(&event) | ||
| return nil | ||
|
|
@@ -448,9 +349,3 @@ func (m *p2pModule) getMultiaddr() (multiaddr.Multiaddr, error) { | |
| "%s:%d", m.cfg.Hostname, m.cfg.Port, | ||
| )) | ||
| } | ||
|
|
||
| // newReadStreamDeadline returns a future deadline | ||
| // based on the read stream timeout duration. | ||
| func newReadStreamDeadline() time.Time { | ||
| return time.Now().Add(readStreamTimeout) | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,31 @@ | ||
| package raintree | ||
|
|
||
| import ( | ||
| libp2pNetwork "github.com/libp2p/go-libp2p/core/network" | ||
|
|
||
| "github.com/pokt-network/pocket/logger" | ||
| "github.com/pokt-network/pocket/p2p/utils" | ||
| ) | ||
|
|
||
| // logStream logs the incoming stream and its scope stats | ||
| func (rtr *rainTreeRouter) logStream(stream libp2pNetwork.Stream) { | ||
| rtr.logStreamScopeStats(stream) | ||
|
|
||
| remotePeer, err := utils.PeerFromLibp2pStream(stream) | ||
| if err != nil { | ||
| rtr.logger.Debug().Err(err).Msg("getting remote remotePeer") | ||
| } else { | ||
| utils.LogIncomingMsg(rtr.logger, rtr.getHostname(), remotePeer) | ||
| } | ||
| } | ||
|
|
||
| // logStreamScopeStats logs the incoming stream's scope stats | ||
| // (see: https://pkg.go.dev/github.com/libp2p/[email protected]/core/network#StreamScope) | ||
| func (rtr *rainTreeRouter) logStreamScopeStats(stream libp2pNetwork.Stream) { | ||
| if err := utils.LogScopeStatFactory( | ||
| &logger.Global.Logger, | ||
| "stream scope (read-side)", | ||
| )(stream.Scope()); err != nil { | ||
| rtr.logger.Debug().Err(err).Msg("logging stream scope stats") | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.