feat: peer sharing protocol

This implementation isn't entirely finished, but there's not yet a
released version of cardano-node supporting this protocol to test
against

Fixes #14

Author: agaffney

.gitignore

@@ -7,6 +7,7 @@
 /gouroboros
 /tx-monitor
 /block-fetch
+/peer-sharing
 
 # Test binary, built with `go test -c`
 *.test

cmd/peer-sharing/main.go

@@ -0,0 +1,59 @@
+// Copyright 2023 Blink Labs, LLC.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"fmt"
+	"os"
+
+	ouroboros "github.com/blinklabs-io/gouroboros"
+	"github.com/blinklabs-io/gouroboros/cmd/common"
+)
+
+func main() {
+	// Parse commandline
+	f := common.NewGlobalFlags()
+	f.Parse()
+	// Create connection
+	conn := common.CreateClientConnection(f)
+	errorChan := make(chan error)
+	go func() {
+		for {
+			err := <-errorChan
+			fmt.Printf("ERROR(async): %s\n", err)
+			os.Exit(1)
+		}
+	}()
+	o, err := ouroboros.New(
+		ouroboros.WithConnection(conn),
+		ouroboros.WithNetworkMagic(uint32(f.NetworkMagic)),
+		ouroboros.WithErrorChan(errorChan),
+		ouroboros.WithNodeToNode(f.NtnProto),
+		ouroboros.WithKeepAlive(true),
+	)
+	if err != nil {
+		fmt.Printf("ERROR: %s\n", err)
+		os.Exit(1)
+	}
+	o.PeerSharing().Client.Start()
+
+	peers, err := o.PeerSharing().Client.GetPeers(10)
+	if err != nil {
+		fmt.Printf("ERROR: %s\n", err)
+		os.Exit(1)
+	}
+
+	fmt.Printf("peers = %#v\n", peers)
+}

connection.go

@@ -39,6 +39,7 @@ import (
 	"github.com/blinklabs-io/gouroboros/protocol/localstatequery"
 	"github.com/blinklabs-io/gouroboros/protocol/localtxmonitor"
 	"github.com/blinklabs-io/gouroboros/protocol/localtxsubmission"
+	"github.com/blinklabs-io/gouroboros/protocol/peersharing"
 	"github.com/blinklabs-io/gouroboros/protocol/txsubmission"
 )
 
@@ -59,19 +60,21 @@ type Connection struct {
 	delayMuxerStart       bool
 	fullDuplex            bool
 	// Mini-protocols
-	handshake               *handshake.Handshake
-	chainSync               *chainsync.ChainSync
-	chainSyncConfig         *chainsync.Config
 	blockFetch              *blockfetch.BlockFetch
 	blockFetchConfig        *blockfetch.Config
+	chainSync               *chainsync.ChainSync
+	chainSyncConfig         *chainsync.Config
+	handshake               *handshake.Handshake
 	keepAlive               *keepalive.KeepAlive
 	keepAliveConfig         *keepalive.Config
+	localStateQuery         *localstatequery.LocalStateQuery
+	localStateQueryConfig   *localstatequery.Config
 	localTxMonitor          *localtxmonitor.LocalTxMonitor
 	localTxMonitorConfig    *localtxmonitor.Config
 	localTxSubmission       *localtxsubmission.LocalTxSubmission
 	localTxSubmissionConfig *localtxsubmission.Config
-	localStateQuery         *localstatequery.LocalStateQuery
-	localStateQueryConfig   *localstatequery.Config
+	peerSharing             *peersharing.PeerSharing
+	peerSharingConfig       *peersharing.Config
 	txSubmission            *txsubmission.TxSubmission
 	txSubmissionConfig      *txsubmission.Config
 }
@@ -165,16 +168,16 @@ func (c *Connection) Close() error {
 	return err
 }
 
-// ChainSync returns the chain-sync protocol handler
-func (c *Connection) ChainSync() *chainsync.ChainSync {
-	return c.chainSync
-}
-
 // BlockFetch returns the block-fetch protocol handler
 func (c *Connection) BlockFetch() *blockfetch.BlockFetch {
 	return c.blockFetch
 }
 
+// ChainSync returns the chain-sync protocol handler
+func (c *Connection) ChainSync() *chainsync.ChainSync {
+	return c.chainSync
+}
+
 // Handshake returns the handshake protocol handler
 func (c *Connection) Handshake() *handshake.Handshake {
 	return c.handshake
@@ -185,6 +188,11 @@ func (c *Connection) KeepAlive() *keepalive.KeepAlive {
 	return c.keepAlive
 }
 
+// LocalStateQuery returns the local-state-query protocol handler
+func (c *Connection) LocalStateQuery() *localstatequery.LocalStateQuery {
+	return c.localStateQuery
+}
+
 // LocalTxMonitor returns the local-tx-monitor protocol handler
 func (c *Connection) LocalTxMonitor() *localtxmonitor.LocalTxMonitor {
 	return c.localTxMonitor
@@ -195,9 +203,9 @@ func (c *Connection) LocalTxSubmission() *localtxsubmission.LocalTxSubmission {
 	return c.localTxSubmission
 }
 
-// LocalStateQuery returns the local-state-query protocol handler
-func (c *Connection) LocalStateQuery() *localstatequery.LocalStateQuery {
-	return c.localStateQuery
+// PeerSharing returns the peer-sharing protocol handler
+func (c *Connection) PeerSharing() *peersharing.PeerSharing {
+	return c.peerSharing
 }
 
 // TxSubmission returns the tx-submission protocol handler
@@ -320,6 +328,9 @@ func (c *Connection) setupConnection() error {
 				c.keepAlive.Client.Start()
 			}
 		}
+		if versionNtN.EnablePeerSharingProtocol {
+			c.peerSharing = peersharing.New(protoOptions, c.peerSharingConfig)
+		}
 	} else {
 		versionNtC := GetProtocolVersionNtC(handshakeVersion)
 		protoOptions.Mode = protocol.ProtocolModeNodeToClient

connection_options.go

@@ -22,6 +22,7 @@ import (
 	"github.com/blinklabs-io/gouroboros/protocol/keepalive"
 	"github.com/blinklabs-io/gouroboros/protocol/localstatequery"
 	"github.com/blinklabs-io/gouroboros/protocol/localtxsubmission"
+	"github.com/blinklabs-io/gouroboros/protocol/peersharing"
 	"github.com/blinklabs-io/gouroboros/protocol/txsubmission"
 )
 
@@ -128,6 +129,13 @@ func WithLocalTxSubmissionConfig(cfg localtxsubmission.Config) ConnectionOptionF
 	}
 }
 
+// WithPeerSharingConfig specifies PeerSharing protocol config
+func WithPeerSharingConfig(cfg peersharing.Config) ConnectionOptionFunc {
+	return func(c *Connection) {
+		c.peerSharingConfig = &cfg
+	}
+}
+
 // WithTxSubmissionConfig specifies TxSubmission protocol config
 func WithTxSubmissionConfig(cfg txsubmission.Config) ConnectionOptionFunc {
 	return func(c *Connection) {

protocol/handshake/client.go

@@ -72,6 +72,13 @@ func (c *Client) Start() {
 		}
 		for _, version := range c.config.ProtocolVersions {
 			if c.Mode() == protocol.ProtocolModeNodeToNode {
+				// NOTE: it seems that protocol version 11 is still in flux, so we disable for now
+				/*
+					if version >= 11 {
+						// TODO: make peer sharing mode configurable
+						versionMap[version] = []interface{}{c.config.NetworkMagic, diffusionMode, PeerSharingModePeerSharingPrivate}
+					} else {
+				*/
 				versionMap[version] = []interface{}{c.config.NetworkMagic, diffusionMode}
 			} else {
 				versionMap[version] = c.config.NetworkMagic

protocol/handshake/handshake.go

@@ -33,6 +33,13 @@ const (
 	DiffusionModeInitiatorAndResponder = true
 )
 
+// Peer sharing modes
+const (
+	PeerSharingModeNoPeerSharing      = 0
+	PeerSharingModePeerSharingPublic  = 1
+	PeerSharingModePeerSharingPrivate = 2
+)
+
 var (
 	statePropose = protocol.NewState(1, "Propose")
 	stateConfirm = protocol.NewState(2, "Confirm")

protocol/peersharing/client.go

@@ -0,0 +1,87 @@
+// Copyright 2023 Blink Labs, LLC.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package peersharing
+
+import (
+	"fmt"
+
+	"github.com/blinklabs-io/gouroboros/protocol"
+)
+
+// Client implements the PeerSharing client
+type Client struct {
+	*protocol.Protocol
+	config         *Config
+	sharePeersChan chan []interface{}
+}
+
+// NewClient returns a new PeerSharing client object
+func NewClient(protoOptions protocol.ProtocolOptions, cfg *Config) *Client {
+	if cfg == nil {
+		tmpCfg := NewConfig()
+		cfg = &tmpCfg
+	}
+	c := &Client{
+		config:         cfg,
+		sharePeersChan: make(chan []interface{}),
+	}
+	// Update state map with timeout
+	stateMap := StateMap.Copy()
+	if entry, ok := stateMap[stateBusy]; ok {
+		entry.Timeout = c.config.Timeout
+		stateMap[stateBusy] = entry
+	}
+	// Configure underlying Protocol
+	protoConfig := protocol.ProtocolConfig{
+		Name:                ProtocolName,
+		ProtocolId:          ProtocolId,
+		Muxer:               protoOptions.Muxer,
+		ErrorChan:           protoOptions.ErrorChan,
+		Mode:                protoOptions.Mode,
+		Role:                protocol.ProtocolRoleClient,
+		MessageHandlerFunc:  c.handleMessage,
+		MessageFromCborFunc: NewMsgFromCbor,
+		StateMap:            stateMap,
+		InitialState:        stateIdle,
+	}
+	c.Protocol = protocol.New(protoConfig)
+	return c
+}
+
+func (c *Client) GetPeers(amount uint8) ([]interface{}, error) {
+	msg := NewMsgShareRequest(amount)
+	if err := c.SendMessage(msg); err != nil {
+		return nil, err
+	}
+	peers := <-c.sharePeersChan
+	return peers, nil
+}
+
+func (c *Client) handleMessage(msg protocol.Message, isResponse bool) error {
+	var err error
+	switch msg.Type() {
+	case MessageTypeSharePeers:
+		err = c.handleSharePeers(msg)
+	default:
+		err = fmt.Errorf("%s: received unexpected message type %d", ProtocolName, msg.Type())
+	}
+	return err
+}
+
+func (c *Client) handleSharePeers(msg protocol.Message) error {
+	msgSharePeers := msg.(*MsgSharePeers)
+	c.sharePeersChan <- msgSharePeers.PeerAddresses
+	return nil
+}