feat: initial LeoisFetch protocol

Signed-off-by: Aurora Gaffney <aurora@blinklabs.io>

Author: agaffney

connection.go

@@ -39,6 +39,7 @@ import (
 "github.com/blinklabs-io/gouroboros/protocol/chainsync"
 "github.com/blinklabs-io/gouroboros/protocol/handshake"
 "github.com/blinklabs-io/gouroboros/protocol/keepalive"
+"github.com/blinklabs-io/gouroboros/protocol/leiosfetch"
 "github.com/blinklabs-io/gouroboros/protocol/leiosnotify"
 "github.com/blinklabs-io/gouroboros/protocol/localstatequery"
 "github.com/blinklabs-io/gouroboros/protocol/localtxmonitor"
@@ -85,6 +86,8 @@ type Connection struct {
 handshake *handshake.Handshake
 keepAlive *keepalive.KeepAlive
 keepAliveConfig *keepalive.Config
+leiosFetch *leiosfetch.LeiosFetch
+leiosFetchConfig *leiosfetch.Config
 leiosNotify *leiosnotify.LeiosNotify
 leiosNotifyConfig *leiosnotify.Config
 localStateQuery *localstatequery.LocalStateQuery
@@ -209,6 +212,11 @@ func (c *Connection) KeepAlive() *keepalive.KeepAlive {
 return c.keepAlive
 }
 
+// LeiosFetch returns the leios-fetch protocol handler
+func (c *Connection) LeiosFetch() *leiosfetch.LeiosFetch {
+return c.leiosFetch
+}
+
 // LeiosNotify returns the leios-notify protocol handler
 func (c *Connection) LeiosNotify() *leiosnotify.LeiosNotify {
 return c.leiosNotify
@@ -405,6 +413,7 @@ func (c *Connection) setupConnection() error {
 c.peerSharing = peersharing.New(protoOptions, c.peerSharingConfig)
 }
 c.leiosNotify = leiosnotify.New(protoOptions, c.leiosNotifyConfig)
+c.leiosFetch = leiosfetch.New(protoOptions, c.leiosFetchConfig)
 // Start protocols
 if !c.delayProtocolStart {
 if (c.fullDuplex && handshakeFullDuplex) || !c.server {
@@ -418,6 +427,7 @@ func (c *Connection) setupConnection() error {
 c.peerSharing.Client.Start()
 }
 c.leiosNotify.Client.Start()
+c.leiosFetch.Client.Start()
 }
 if (c.fullDuplex && handshakeFullDuplex) || c.server {
 c.blockFetch.Server.Start()
@@ -430,6 +440,7 @@ func (c *Connection) setupConnection() error {
 c.peerSharing.Server.Start()
 }
 c.leiosNotify.Server.Start()
+c.leiosFetch.Server.Start()
 }
 }
 } else {

protocol/leiosfetch/client.go

@@ -0,0 +1,230 @@
+// Copyright 2025 Blink Labs Software
+//
+// 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 leiosfetch
+
+import (
+"fmt"
+"sync"
+
+"github.com/blinklabs-io/gouroboros/protocol"
+pcommon "github.com/blinklabs-io/gouroboros/protocol/common"
+)
+
+type Client struct {
+*protocol.Protocol
+config *Config
+callbackContext CallbackContext
+onceStart sync.Once
+onceStop sync.Once
+blockResultChan chan protocol.Message
+blockTxsResultChan chan protocol.Message
+votesResultChan chan protocol.Message
+blockRangeResultChan chan protocol.Message
+}
+
+func NewClient(protoOptions protocol.ProtocolOptions, cfg *Config) *Client {
+if cfg == nil {
+tmpCfg := NewConfig()
+cfg = &tmpCfg
+}
+c := &Client{
+config: cfg,
+blockResultChan: make(chan protocol.Message),
+blockTxsResultChan: make(chan protocol.Message),
+votesResultChan: make(chan protocol.Message),
+blockRangeResultChan: make(chan protocol.Message),
+}
+c.callbackContext = CallbackContext{
+Client: c,
+ConnectionId: protoOptions.ConnectionId,
+}
+// Update state map with timeout
+stateMap := StateMap.Copy()
+if entry, ok := stateMap[StateBlock]; ok {
+entry.Timeout = c.config.Timeout
+stateMap[StateBlock] = entry
+}
+if entry, ok := stateMap[StateBlockTxs]; ok {
+entry.Timeout = c.config.Timeout
+stateMap[StateBlockTxs] = entry
+}
+if entry, ok := stateMap[StateVotes]; ok {
+entry.Timeout = c.config.Timeout
+stateMap[StateVotes] = entry
+}
+if entry, ok := stateMap[StateBlockRange]; ok {
+entry.Timeout = c.config.Timeout
+stateMap[StateBlockRange] = entry
+}
+// Configure underlying Protocol
+protoConfig := protocol.ProtocolConfig{
+Name: ProtocolName,
+ProtocolId: ProtocolId,
+Muxer: protoOptions.Muxer,
+Logger: protoOptions.Logger,
+ErrorChan: protoOptions.ErrorChan,
+Mode: protoOptions.Mode,
+Role: protocol.ProtocolRoleClient,
+MessageHandlerFunc: c.messageHandler,
+MessageFromCborFunc: NewMsgFromCbor,
+StateMap: stateMap,
+InitialState: StateIdle,
+}
+c.Protocol = protocol.New(protoConfig)
+return c
+}
+
+func (c *Client) Start() {
+c.onceStart.Do(func() {
+c.Protocol.Logger().
+Debug("starting client protocol",
+"component", "network",
+"protocol", ProtocolName,
+"connection_id", c.callbackContext.ConnectionId.String(),
+)
+c.Protocol.Start()
+// Start goroutine to cleanup resources on protocol shutdown
+go func() {
+<-c.DoneChan()
+close(c.blockResultChan)
+close(c.blockTxsResultChan)
+close(c.votesResultChan)
+close(c.blockRangeResultChan)
+}()
+})
+}
+
+func (c *Client) Stop() error {
+var err error
+c.onceStop.Do(func() {
+c.Protocol.Logger().
+Debug("stopping client protocol",
+"component", "network",
+"protocol", ProtocolName,
+"connection_id", c.callbackContext.ConnectionId.String(),
+)
+msg := NewMsgDone()
+err = c.SendMessage(msg)
+})
+return err
+}
+
+// BlockRequest fetches the requested EB identified by the slot and Leios hash
+func (c *Client) BlockRequest(slot uint64, hash []byte) (protocol.Message, error) {
+msg := NewMsgBlockRequest(slot, hash)
+if err := c.SendMessage(msg); err != nil {
+return nil, err
+}
+resp, ok := <-c.blockResultChan
+if !ok {
+return nil, protocol.ErrProtocolShuttingDown
+}
+return resp, nil
+}
+
+// BlockTxsRequest fetches the requested TXs identified by the slot, Leios hash, and TX bitmap
+func (c *Client) BlockTxsRequest(slot uint64, hash []byte, txBitmap [8]byte) (protocol.Message, error) {
+msg := NewMsgBlockTxsRequest(slot, hash, txBitmap)
+if err := c.SendMessage(msg); err != nil {
+return nil, err
+}
+resp, ok := <-c.blockTxsResultChan
+if !ok {
+return nil, protocol.ErrProtocolShuttingDown
+}
+return resp, nil
+}
+
+// VotesRequest fetches the requested votes
+func (c *Client) VotesRequest(voteIds []MsgVotesRequestVoteId) (protocol.Message, error) {
+msg := NewMsgVotesRequest(voteIds)
+if err := c.SendMessage(msg); err != nil {
+return nil, err
+}
+resp, ok := <-c.votesResultChan
+if !ok {
+return nil, protocol.ErrProtocolShuttingDown
+}
+return resp, nil
+}
+
+// BlockRangeRequest fetches a range of EBs and their TXs that are certified by RBs within the provided range.
+// This function will block until all EBs and TXs in the requested range have been received
+func (c *Client) BlockRangeRequest(start pcommon.Point, end pcommon.Point) ([]protocol.Message, error) {
+msg := NewMsgBlockRangeRequest(start, end)
+if err := c.SendMessage(msg); err != nil {
+return nil, err
+}
+ret := make([]protocol.Message, 0, 20)
+for {
+resp, ok := <-c.blockResultChan
+if !ok {
+return nil, protocol.ErrProtocolShuttingDown
+}
+ret = append(ret, resp)
+if _, ok := resp.(*MsgLastBlockAndTxsInRange); ok {
+break
+}
+}
+return ret, nil
+}
+
+func (c *Client) messageHandler(msg protocol.Message) error {
+var err error
+switch msg.Type() {
+case MessageTypeBlock:
+err = c.handleBlock(msg)
+case MessageTypeBlockTxs:
+err = c.handleBlockTxs(msg)
+case MessageTypeVotes:
+err = c.handleVotes(msg)
+case MessageTypeNextBlockAndTxsInRange:
+err = c.handleNextBlockAndTxsInRange(msg)
+case MessageTypeLastBlockAndTxsInRange:
+err = c.handleLastBlockAndTxsInRange(msg)
+default:
+err = fmt.Errorf(
+"%s: received unexpected message type %d",
+ProtocolName,
+msg.Type(),
+)
+}
+return err
+}
+
+func (c *Client) handleBlock(msg protocol.Message) error {
+c.blockResultChan <- msg
+return nil
+}
+
+func (c *Client) handleBlockTxs(msg protocol.Message) error {
+c.blockTxsResultChan <- msg
+return nil
+}
+
+func (c *Client) handleVotes(msg protocol.Message) error {
+c.votesResultChan <- msg
+return nil
+}
+
+func (c *Client) handleNextBlockAndTxsInRange(msg protocol.Message) error {
+c.blockRangeResultChan <- msg
+return nil
+}
+
+func (c *Client) handleLastBlockAndTxsInRange(msg protocol.Message) error {
+c.blockRangeResultChan <- msg
+return nil
+}