feat(parachain): send requests from subsystems

dot/network/service.go

@@ -593,7 +593,7 @@ func (s *Service) SendMessage(to peer.ID, msg NotificationsMessage) error {
 }
 
 func (s *Service) GetRequestResponseProtocol(subprotocol string, requestTimeout time.Duration,
-	maxResponseSize uint64) *RequestResponseProtocol {
+	maxResponseSize uint64) RequestMaker {
 
 	protocolID := s.host.protocolID + protocol.ID(subprotocol)
 	return &RequestResponseProtocol{

dot/parachain/collator-protocol/validator_side.go

@@ -559,7 +559,7 @@ type Network interface {
 		maxSize uint64,
 	) error
 	GetRequestResponseProtocol(subprotocol string, requestTimeout time.Duration,
-		maxResponseSize uint64) *network.RequestResponseProtocol
+		maxResponseSize uint64) network.RequestMaker
 }
 
 type CollationEvent struct {
@@ -574,7 +574,7 @@ type CollatorProtocolValidatorSide struct {
 	SubSystemToOverseer chan<- any
 	unfetchedCollation  chan UnfetchedCollation
 
-	collationFetchingReqResProtocol *network.RequestResponseProtocol
+	collationFetchingReqResProtocol network.RequestMaker
 
 	fetchedCollations []parachaintypes.Collation
 	// track all active collators and their data

dot/parachain/network-bridge/interface.go

@@ -27,7 +27,7 @@ type Network interface {
 		maxSize uint64,
 	) error
 	GetRequestResponseProtocol(subprotocol string, requestTimeout time.Duration,
-		maxResponseSize uint64) *network.RequestResponseProtocol
+		maxResponseSize uint64) network.RequestMaker
 	ReportPeer(change peerset.ReputationChange, p peer.ID)
 	DisconnectPeer(setID int, p peer.ID)
 	GetNetworkEventsChannel() chan *network.NetworkEventInfo

dot/parachain/chunk_fetching.go → dot/parachain/network-bridge/messages/chunk_fetching.go

@@ -1,11 +1,13 @@
 // Copyright 2023 ChainSafe Systems (ON)
 // SPDX-License-Identifier: LGPL-3.0-only
 
-package parachain
+package messages
 
 import (
 	"fmt"
 
+	"github.com/ChainSafe/gossamer/dot/network"
+
 	parachaintypes "github.com/ChainSafe/gossamer/dot/parachain/types"
 	"github.com/ChainSafe/gossamer/pkg/scale"
 )
@@ -24,6 +26,16 @@ func (c ChunkFetchingRequest) Encode() ([]byte, error) {
 	return scale.Marshal(c)
 }
 
+// Protocol returns the sub-protocol ID for this message
+func (c ChunkFetchingRequest) Protocol() ReqProtocolName {
+	return ChunkFetchingV1
+}
+
+// Response returns an instance of the response type for this message, for the purpose of decoding into it.
+func (c ChunkFetchingRequest) Response() network.ResponseMessage {
+	return &ChunkFetchingResponse{}
+}
+
 type ChunkFetchingResponseValues interface {
 	ChunkResponse | NoSuchChunk
 }

dot/parachain/chunk_fetching_test.go → dot/parachain/network-bridge/messages/chunk_fetching_test.go

@@ -1,7 +1,7 @@
 // Copyright 2023 ChainSafe Systems (ON)
 // SPDX-License-Identifier: LGPL-3.0-only
 
-package parachain
+package messages
 
 import (
 	"testing"

dot/parachain/network-bridge/messages/request_response_protocols.go

@@ -0,0 +1,68 @@
+package messages
+
+import (
+	"github.com/ChainSafe/gossamer/dot/network"
+	"github.com/libp2p/go-libp2p/core/peer"
+)
+
+type ReqProtocolName uint
+
+const (
+	ChunkFetchingV1 ReqProtocolName = iota
+	CollationFetchingV1
+	PoVFetchingV1
+	AvailableDataFetchingV1
+	StatementFetchingV1
+	DisputeSendingV1
+)
+
+func (n ReqProtocolName) String() string {
+	switch n {
+	case ChunkFetchingV1:
+		return "req_chunk/1"
+	case CollationFetchingV1:
+		return "req_collation/1"
+	case PoVFetchingV1:
+		return "req_pov/1"
+	case AvailableDataFetchingV1:
+		return "req_available_data/1"
+	case StatementFetchingV1:
+		return "req_statement/1"
+	case DisputeSendingV1:
+		return "send_dispute/1"
+	default:
+		panic("unknown protocol")
+	}
+}
+
+// ReqProtocolMessage is a network message that can be sent over a request response protocol.
+type ReqProtocolMessage interface {
+	network.Message
+	// Response returns an instance of the response type for this message, for the purpose of decoding into it.
+	Response() network.ResponseMessage
+	Protocol() ReqProtocolName
+}
+
+// ReqRespResult is the result of sending a request over a request response protocol. It contains either a response
+// message or an error.
+type ReqRespResult struct {
+	Response network.ResponseMessage
+	Error    error
+}
+
+// OutgoingRequest contains all data required to send a request over a request response protocol and receive the result.
+type OutgoingRequest struct {
+	Recipient peer.ID // TODO use a type that can contain either a peer ID or an authority ID
+	Payload   ReqProtocolMessage
+	Result    chan ReqRespResult
+}
+
+func NewOutgoingRequest(recipient peer.ID, payload ReqProtocolMessage) *OutgoingRequest {
+	result := make(chan ReqRespResult, 1)
+
+	return &OutgoingRequest{
+		Recipient: recipient,
+		Payload:   payload,
+		Result:    result,
+	}
+}

dot/parachain/network-bridge/messages/tx_overseer_messages.go

@@ -63,3 +63,16 @@ type ConnectToValidators struct {
 	// authority discovery has Failed to resolve.
 	Failed chan<- uint
 }
+
+type IfDisconnectedBehavior int
+
+const (
+	TryConnect     IfDisconnectedBehavior = iota
+	ImmediateError                        // TODO not implemented
+)
+
+// SendRequests is a subsystem message for sending requests over a request response protocol.
+type SendRequests struct {
+	Requests       []*OutgoingRequest
+	IfDisconnected IfDisconnectedBehavior
+}

dot/parachain/network-bridge/sender.go

@@ -6,6 +6,7 @@ package networkbridge
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/ChainSafe/gossamer/dot/network"
 	networkbridgemessages "github.com/ChainSafe/gossamer/dot/parachain/network-bridge/messages"
@@ -92,7 +93,9 @@ func (nbs *NetworkBridgeSender) processMessage(msg any) error {
 				return fmt.Errorf("sending message: %w", err)
 			}
 		}
-		// TODO: add ConnectTOResolvedValidators, SendRequests
+	case networkbridgemessages.SendRequests:
+		nbs.sendRequests(msg.Requests, msg.IfDisconnected)
+		// TODO: add ConnectTOResolvedValidators
 	case networkbridgemessages.ConnectToValidators:
 		// TODO
 	case networkbridgemessages.ReportPeer:
@@ -104,3 +107,33 @@ func (nbs *NetworkBridgeSender) processMessage(msg any) error {
 
 	return nil
 }
+
+const requestTimeout = 200 * time.Millisecond // TODO is this reasonable?
+
+// PoV is probably the largest message and is currently set at 5MB, but will likely be increased to 10MB in the future.
+// see: https://github.com/paritytech/polkadot-sdk/issues/5334
+// Maybe message types should have a MaxSize() method instead of using the same value for all messages.
+const maxResponseSize uint64 = 5 * 1024 * 1024
+
+func (nbs *NetworkBridgeSender) sendRequests(
+	requests []*networkbridgemessages.OutgoingRequest,
+	ifDisconnected networkbridgemessages.IfDisconnectedBehavior, //nolint:unparam
+) {
+	for _, request := range requests {
+		protoID := request.Payload.Protocol().String()
+		protocol := nbs.net.GetRequestResponseProtocol(protoID, requestTimeout, maxResponseSize)
+		response := request.Payload.Response()
+		result := networkbridgemessages.ReqRespResult{}
+
+		// TODO This should probably be done on a goroutine. Unclear how to deal with cancellation/shutdown though.
+		err := protocol.Do(request.Recipient, request.Payload, response)
+		if err != nil {
+			result.Error = err
+		} else {
+			result.Response = response
+		}
+
+		request.Result <- result
+		close(request.Result)
+	}
+}