(resourcesCtx context.Context, chainID *big.Int, lg log.Logger, setup SetupP2P, storageManager *ethstorage.StorageManager, db ethdb.Database, m metrics.Metricer, feed *event.Feed)
| 67 | } |
| 68 | |
| 69 | func (n *NodeP2P) init(resourcesCtx context.Context, chainID *big.Int, lg log.Logger, setup SetupP2P, |
| 70 | storageManager *ethstorage.StorageManager, db ethdb.Database, m metrics.Metricer, feed *event.Feed) error { |
| 71 | bwc := p2pmetrics.NewBandwidthCounter() |
| 72 | n.storageManager = storageManager |
| 73 | n.lg = lg |
| 74 | n.resCtx = resourcesCtx |
| 75 | |
| 76 | var err error |
| 77 | // nil if disabled. |
| 78 | n.host, err = setup.Host(lg, bwc) |
| 79 | if err != nil { |
| 80 | if n.dv5Udp != nil { |
| 81 | n.dv5Udp.Close() |
| 82 | } |
| 83 | return fmt.Errorf("failed to start p2p host: %w", err) |
| 84 | } |
| 85 | |
| 86 | if n.host != nil { |
| 87 | // Enable extra features, if any. During testing we don't setup the most advanced host all the time. |
| 88 | if extra, ok := n.host.(ExtraHostFeatures); ok { |
| 89 | n.gater = extra.ConnectionGater() |
| 90 | n.connMgr = extra.ConnectionManager() |
| 91 | } |
| 92 | |
| 93 | // Activate the P2P req-resp sync |
| 94 | n.syncCl = protocol.NewSyncClient(lg, chainID, n.host.NewStream, storageManager, setup.SyncerParams(), db, m, feed) |
| 95 | n.host.Network().Notify(&network.NotifyBundle{ |
| 96 | ConnectedF: func(nw network.Network, conn network.Conn) { |
| 97 | var ( |
| 98 | shards map[common.Address][]uint64 |
| 99 | remotePeerId = conn.RemotePeer() |
| 100 | ) |
| 101 | css, err := n.Host().Peerstore().Get(remotePeerId, protocol.EthStorageENRKey) |
| 102 | if err != nil { |
| 103 | // For nodes new to the ethstorage network that dial nodes which do not contain |
| 104 | // the new node's enr, the nodes do not know its shard list from enr, so it needs to call |
| 105 | // n.RequestShardList to fetch the shard list of the new node. |
| 106 | remoteShardList, e := n.RequestShardList(remotePeerId) |
| 107 | if e != nil && len(n.host.Peerstore().Addrs(remotePeerId)) == 0 { |
| 108 | // As the remote node host may enable NATService, which will create a new connection with another |
| 109 | // peer id and its Addrs will not be set to local host's Peerstore. So if len of peer Addrs is 0 and |
| 110 | // cannot get the remote node's shard list, then ignore this connection. |
| 111 | lg.Debug("No addresses to get shard list, return without close conn", "peer", n.host.ID(), "remote peer", |
| 112 | remotePeerId, "Direction", conn.Stat().Direction, "remote address", conn.RemoteMultiaddr().String(), "error", e.Error()) |
| 113 | return |
| 114 | } else if e != nil { |
| 115 | lg.Debug("Get remote shard list fail", "peer", remotePeerId, "Direction", conn.Stat().Direction, |
| 116 | "remote address", conn.RemoteMultiaddr().String(), "err", e.Error()) |
| 117 | conn.Close() |
| 118 | return |
| 119 | } |
| 120 | lg.Debug("Get remote shard list success", "peer", remotePeerId, "shards", remoteShardList, |
| 121 | "Direction", conn.Stat().Direction, "remote address", conn.RemoteMultiaddr().String()) |
| 122 | n.Host().Peerstore().Put(remotePeerId, protocol.EthStorageENRKey, remoteShardList) |
| 123 | shards = protocol.ConvertToShardList(remoteShardList) |
| 124 | } else { |
| 125 | shards = protocol.ConvertToShardList(css.([]*protocol.ContractShards)) |
| 126 | } |
no test coverage detected