LBRYCommunity/lbcd
1
0
Fork 0
Code Issues 17 Pull requests 4 Projects Releases 28 Packages Wiki Activity

server: Ensure callbacks use the server peer.

Browse source 
This modifies the signatures of all serverPeer callbacks that are
provided as peer.Listeners to use _ for the first parameter name which
ensures the passed peer can't be used within the function and updates
all references to the server peer.

This helps ensure any overridden methods that might be defined on a
serverPeer will be invoked where directly calling methods on the passed
peer would not.

Also, while here, add a comment to the OnFeeFilter function.
This commit is contained in:
Dave Collins 2016-11-03 23:58:35 -05:00
parent b65881c137
commit df33d4340e
No known key found for this signature in database
GPG key ID: B8904D9D9C93D1F2
1 changed files with 49 additions and 45 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

94
server.go
View file

@ -314,10 +314,10 @@ func (sp *serverPeer) addBanScore(persistent, transient uint32, reason string) {
// OnVersion is invoked when a peer receives a version bitcoin message // OnVersion is invoked when a peer receives a version bitcoin message
// and is used to negotiate the protocol version details as well as kick start // and is used to negotiate the protocol version details as well as kick start
// the communications. // the communications.
func (sp *serverPeer) OnVersion(p *peer.Peer, msg *wire.MsgVersion) { func (sp *serverPeer) OnVersion(_ *peer.Peer, msg *wire.MsgVersion) {
// Add the remote peer time as a sample for creating an offset against // Add the remote peer time as a sample for creating an offset against
// the local clock to keep the network time in sync. // the local clock to keep the network time in sync.
sp.server.timeSource.AddTimeSample(p.Addr(), msg.Timestamp) sp.server.timeSource.AddTimeSample(sp.Addr(), msg.Timestamp)
// Signal the block manager this peer is a new sync candidate. // Signal the block manager this peer is a new sync candidate.
sp.server.blockManager.NewPeer(sp) sp.server.blockManager.NewPeer(sp)
@ -334,12 +334,12 @@ func (sp *serverPeer) OnVersion(p *peer.Peer, msg *wire.MsgVersion) {
if !cfg.SimNet { if !cfg.SimNet {
addrManager := sp.server.addrManager addrManager := sp.server.addrManager
// Outbound connections. // Outbound connections.
if !p.Inbound() { if !sp.Inbound() {
// TODO(davec): Only do this if not doing the initial block // TODO(davec): Only do this if not doing the initial block
// download and the local address is routable. // download and the local address is routable.
if !cfg.DisableListen /* && isCurrent? */ { if !cfg.DisableListen /* && isCurrent? */ {
// Get address that best matches. // Get address that best matches.
lna := addrManager.GetBestLocalAddress(p.NA()) lna := addrManager.GetBestLocalAddress(sp.NA())
if addrmgr.IsRoutable(lna) { if addrmgr.IsRoutable(lna) {
// Filter addresses the peer already knows about. // Filter addresses the peer already knows about.
addresses := []*wire.NetAddress{lna} addresses := []*wire.NetAddress{lna}
@ -350,14 +350,14 @@ func (sp *serverPeer) OnVersion(p *peer.Peer, msg *wire.MsgVersion) {
// Request known addresses if the server address manager needs // Request known addresses if the server address manager needs
// more and the peer has a protocol version new enough to // more and the peer has a protocol version new enough to
// include a timestamp with addresses. // include a timestamp with addresses.
hasTimestamp := p.ProtocolVersion() >= hasTimestamp := sp.ProtocolVersion() >=
wire.NetAddressTimeVersion wire.NetAddressTimeVersion
if addrManager.NeedMoreAddresses() && hasTimestamp { if addrManager.NeedMoreAddresses() && hasTimestamp {
p.QueueMessage(wire.NewMsgGetAddr(), nil) sp.QueueMessage(wire.NewMsgGetAddr(), nil)
} }
// Mark the address as a known good address. // Mark the address as a known good address.
addrManager.Good(p.NA()) addrManager.Good(sp.NA())
} }
} }
@ -369,7 +369,7 @@ func (sp *serverPeer) OnVersion(p *peer.Peer, msg *wire.MsgVersion) {
// It creates and sends an inventory message with the contents of the memory // It creates and sends an inventory message with the contents of the memory
// pool up to the maximum inventory allowed per message. When the peer has a // pool up to the maximum inventory allowed per message. When the peer has a
// bloom filter loaded, the contents are filtered accordingly. // bloom filter loaded, the contents are filtered accordingly.
func (sp *serverPeer) OnMemPool(p *peer.Peer, msg *wire.MsgMemPool) { func (sp *serverPeer) OnMemPool(_ *peer.Peer, msg *wire.MsgMemPool) {
// A decaying ban score increase is applied to prevent flooding. // A decaying ban score increase is applied to prevent flooding.
// The ban score accumulates and passes the ban threshold if a burst of // The ban score accumulates and passes the ban threshold if a burst of
// mempool messages comes from a peer. The score decays each minute to // mempool messages comes from a peer. The score decays each minute to
@ -400,7 +400,7 @@ func (sp *serverPeer) OnMemPool(p *peer.Peer, msg *wire.MsgMemPool) {
// Send the inventory message if there is anything to send. // Send the inventory message if there is anything to send.
if len(invMsg.InvList) > 0 { if len(invMsg.InvList) > 0 {
p.QueueMessage(invMsg, nil) sp.QueueMessage(invMsg, nil)
} }
} }
@ -408,10 +408,10 @@ func (sp *serverPeer) OnMemPool(p *peer.Peer, msg *wire.MsgMemPool) {
// until the bitcoin transaction has been fully processed. Unlock the block // until the bitcoin transaction has been fully processed. Unlock the block
// handler this does not serialize all transactions through a single thread // handler this does not serialize all transactions through a single thread
// transactions don't rely on the previous one in a linear fashion like blocks. // transactions don't rely on the previous one in a linear fashion like blocks.
func (sp *serverPeer) OnTx(p *peer.Peer, msg *wire.MsgTx) { func (sp *serverPeer) OnTx(_ *peer.Peer, msg *wire.MsgTx) {
if cfg.BlocksOnly { if cfg.BlocksOnly {
peerLog.Tracef("Ignoring tx %v from %v - blocksonly enabled", peerLog.Tracef("Ignoring tx %v from %v - blocksonly enabled",
msg.TxHash(), p) msg.TxHash(), sp)
return return
} }
@ -420,7 +420,7 @@ func (sp *serverPeer) OnTx(p *peer.Peer, msg *wire.MsgTx) {
// methods and things such as hash caching. // methods and things such as hash caching.
tx := btcutil.NewTx(msg) tx := btcutil.NewTx(msg)
iv := wire.NewInvVect(wire.InvTypeTx, tx.Hash()) iv := wire.NewInvVect(wire.InvTypeTx, tx.Hash())
p.AddKnownInventory(iv) sp.AddKnownInventory(iv)
// Queue the transaction up to be handled by the block manager and // Queue the transaction up to be handled by the block manager and
// intentionally block further receives until the transaction is fully // intentionally block further receives until the transaction is fully
@ -433,14 +433,14 @@ func (sp *serverPeer) OnTx(p *peer.Peer, msg *wire.MsgTx) {
// OnBlock is invoked when a peer receives a block bitcoin message. It // OnBlock is invoked when a peer receives a block bitcoin message. It
// blocks until the bitcoin block has been fully processed. // blocks until the bitcoin block has been fully processed.
func (sp *serverPeer) OnBlock(p *peer.Peer, msg *wire.MsgBlock, buf []byte) { func (sp *serverPeer) OnBlock(_ *peer.Peer, msg *wire.MsgBlock, buf []byte) {
// Convert the raw MsgBlock to a btcutil.Block which provides some // Convert the raw MsgBlock to a btcutil.Block which provides some
// convenience methods and things such as hash caching. // convenience methods and things such as hash caching.
block := btcutil.NewBlockFromBlockAndBytes(msg, buf) block := btcutil.NewBlockFromBlockAndBytes(msg, buf)
// Add the block to the known inventory for the peer. // Add the block to the known inventory for the peer.
iv := wire.NewInvVect(wire.InvTypeBlock, block.Hash()) iv := wire.NewInvVect(wire.InvTypeBlock, block.Hash())
p.AddKnownInventory(iv) sp.AddKnownInventory(iv)
// Queue the block up to be handled by the block // Queue the block up to be handled by the block
// manager and intentionally block further receives // manager and intentionally block further receives
@ -461,7 +461,7 @@ func (sp *serverPeer) OnBlock(p *peer.Peer, msg *wire.MsgBlock, buf []byte) {
// used to examine the inventory being advertised by the remote peer and react // used to examine the inventory being advertised by the remote peer and react
// accordingly. We pass the message down to blockmanager which will call // accordingly. We pass the message down to blockmanager which will call
// QueueMessage with any appropriate responses. // QueueMessage with any appropriate responses.
func (sp *serverPeer) OnInv(p *peer.Peer, msg *wire.MsgInv) { func (sp *serverPeer) OnInv(_ *peer.Peer, msg *wire.MsgInv) {
if !cfg.BlocksOnly { if !cfg.BlocksOnly {
if len(msg.InvList) > 0 { if len(msg.InvList) > 0 {
sp.server.blockManager.QueueInv(msg, sp) sp.server.blockManager.QueueInv(msg, sp)
@ -473,11 +473,11 @@ func (sp *serverPeer) OnInv(p *peer.Peer, msg *wire.MsgInv) {
for _, invVect := range msg.InvList { for _, invVect := range msg.InvList {
if invVect.Type == wire.InvTypeTx { if invVect.Type == wire.InvTypeTx {
peerLog.Tracef("Ignoring tx %v in inv from %v -- "+ peerLog.Tracef("Ignoring tx %v in inv from %v -- "+
"blocksonly enabled", invVect.Hash, p) "blocksonly enabled", invVect.Hash, sp)
if p.ProtocolVersion() >= wire.BIP0037Version { if sp.ProtocolVersion() >= wire.BIP0037Version {
peerLog.Infof("Peer %v is announcing "+ peerLog.Infof("Peer %v is announcing "+
"transactions -- disconnecting", p) "transactions -- disconnecting", sp)
p.Disconnect() sp.Disconnect()
return return
} }
continue continue
@ -496,13 +496,13 @@ func (sp *serverPeer) OnInv(p *peer.Peer, msg *wire.MsgInv) {
// OnHeaders is invoked when a peer receives a headers bitcoin // OnHeaders is invoked when a peer receives a headers bitcoin
// message. The message is passed down to the block manager. // message. The message is passed down to the block manager.
func (sp *serverPeer) OnHeaders(p *peer.Peer, msg *wire.MsgHeaders) { func (sp *serverPeer) OnHeaders(_ *peer.Peer, msg *wire.MsgHeaders) {
sp.server.blockManager.QueueHeaders(msg, sp) sp.server.blockManager.QueueHeaders(msg, sp)
} }
// handleGetData is invoked when a peer receives a getdata bitcoin message and // handleGetData is invoked when a peer receives a getdata bitcoin message and
// is used to deliver block and transaction information. // is used to deliver block and transaction information.
func (sp *serverPeer) OnGetData(p *peer.Peer, msg *wire.MsgGetData) { func (sp *serverPeer) OnGetData(_ *peer.Peer, msg *wire.MsgGetData) {
numAdded := 0 numAdded := 0
notFound := wire.NewMsgNotFound() notFound := wire.NewMsgNotFound()
@ -561,7 +561,7 @@ func (sp *serverPeer) OnGetData(p *peer.Peer, msg *wire.MsgGetData) {
waitChan = c waitChan = c
} }
if len(notFound.InvList) != 0 { if len(notFound.InvList) != 0 {
p.QueueMessage(notFound, doneChan) sp.QueueMessage(notFound, doneChan)
} }
// Wait for messages to be sent. We can send quite a lot of data at this // Wait for messages to be sent. We can send quite a lot of data at this
@ -576,7 +576,7 @@ func (sp *serverPeer) OnGetData(p *peer.Peer, msg *wire.MsgGetData) {
// OnGetBlocks is invoked when a peer receives a getblocks bitcoin // OnGetBlocks is invoked when a peer receives a getblocks bitcoin
// message. // message.
func (sp *serverPeer) OnGetBlocks(p *peer.Peer, msg *wire.MsgGetBlocks) { func (sp *serverPeer) OnGetBlocks(_ *peer.Peer, msg *wire.MsgGetBlocks) {
// Return all block hashes to the latest one (up to max per message) if // Return all block hashes to the latest one (up to max per message) if
// no stop hash was specified. // no stop hash was specified.
// Attempt to find the ending index of the stop hash if specified. // Attempt to find the ending index of the stop hash if specified.
@ -636,13 +636,13 @@ func (sp *serverPeer) OnGetBlocks(p *peer.Peer, msg *wire.MsgGetBlocks) {
continueHash := invMsg.InvList[invListLen-1].Hash continueHash := invMsg.InvList[invListLen-1].Hash
sp.continueHash = &continueHash sp.continueHash = &continueHash
} }
p.QueueMessage(invMsg, nil) sp.QueueMessage(invMsg, nil)
} }
} }
// OnGetHeaders is invoked when a peer receives a getheaders bitcoin // OnGetHeaders is invoked when a peer receives a getheaders bitcoin
// message. // message.
func (sp *serverPeer) OnGetHeaders(p *peer.Peer, msg *wire.MsgGetHeaders) { func (sp *serverPeer) OnGetHeaders(_ *peer.Peer, msg *wire.MsgGetHeaders) {
// Ignore getheaders requests if not in sync. // Ignore getheaders requests if not in sync.
if !sp.server.blockManager.IsCurrent() { if !sp.server.blockManager.IsCurrent() {
return return
@ -690,7 +690,7 @@ func (sp *serverPeer) OnGetHeaders(p *peer.Peer, msg *wire.MsgGetHeaders) {
headersMsg := wire.NewMsgHeaders() headersMsg := wire.NewMsgHeaders()
headersMsg.AddBlockHeader(&header) headersMsg.AddBlockHeader(&header)
p.QueueMessage(headersMsg, nil) sp.QueueMessage(headersMsg, nil)
return return
} }
@ -745,7 +745,7 @@ func (sp *serverPeer) OnGetHeaders(p *peer.Peer, msg *wire.MsgGetHeaders) {
return return
} }
p.QueueMessage(headersMsg, nil) sp.QueueMessage(headersMsg, nil)
} }
// enforceNodeBloomFlag disconnects the peer if the server is not configured to // enforceNodeBloomFlag disconnects the peer if the server is not configured to
@ -783,7 +783,11 @@ func (sp *serverPeer) enforceNodeBloomFlag(cmd string) bool {
return true return true
} }
func (sp *serverPeer) OnFeeFilter(p *peer.Peer, msg *wire.MsgFeeFilter) { // OnFeeFilter is invoked when a peer receives a feefilter bitcoin message and
// is used by remote peers to request that no transactions which have a fee rate
// lower than provided value are inventoried to them. The peer will be
// disconnected if an invalid fee filter value is provided.
func (sp *serverPeer) OnFeeFilter(_ *peer.Peer, msg *wire.MsgFeeFilter) {
// Check that the passed minimum fee is a valid amount. // Check that the passed minimum fee is a valid amount.
if msg.MinFee < 0 || msg.MinFee > btcutil.MaxSatoshi { if msg.MinFee < 0 || msg.MinFee > btcutil.MaxSatoshi {
peerLog.Debugf("Peer %v sent an invalid feefilter '%v' -- "+ peerLog.Debugf("Peer %v sent an invalid feefilter '%v' -- "+
@ -799,7 +803,7 @@ func (sp *serverPeer) OnFeeFilter(p *peer.Peer, msg *wire.MsgFeeFilter) {
// message and is used by remote peers to add data to an already loaded bloom // message and is used by remote peers to add data to an already loaded bloom
// filter. The peer will be disconnected if a filter is not loaded when this // filter. The peer will be disconnected if a filter is not loaded when this
// message is received or the server is not configured to allow bloom filters. // message is received or the server is not configured to allow bloom filters.
func (sp *serverPeer) OnFilterAdd(p *peer.Peer, msg *wire.MsgFilterAdd) { func (sp *serverPeer) OnFilterAdd(_ *peer.Peer, msg *wire.MsgFilterAdd) {
// Disconnect and/or ban depending on the node bloom services flag and // Disconnect and/or ban depending on the node bloom services flag and
// negotiated protocol version. // negotiated protocol version.
if !sp.enforceNodeBloomFlag(msg.Command()) { if !sp.enforceNodeBloomFlag(msg.Command()) {
@ -808,8 +812,8 @@ func (sp *serverPeer) OnFilterAdd(p *peer.Peer, msg *wire.MsgFilterAdd) {
if sp.filter.IsLoaded() { if sp.filter.IsLoaded() {
peerLog.Debugf("%s sent a filteradd request with no filter "+ peerLog.Debugf("%s sent a filteradd request with no filter "+
"loaded -- disconnecting", p) "loaded -- disconnecting", sp)
p.Disconnect() sp.Disconnect()
return return
} }
@ -820,7 +824,7 @@ func (sp *serverPeer) OnFilterAdd(p *peer.Peer, msg *wire.MsgFilterAdd) {
// message and is used by remote peers to clear an already loaded bloom filter. // message and is used by remote peers to clear an already loaded bloom filter.
// The peer will be disconnected if a filter is not loaded when this message is // The peer will be disconnected if a filter is not loaded when this message is
// received or the server is not configured to allow bloom filters. // received or the server is not configured to allow bloom filters.
func (sp *serverPeer) OnFilterClear(p *peer.Peer, msg *wire.MsgFilterClear) { func (sp *serverPeer) OnFilterClear(_ *peer.Peer, msg *wire.MsgFilterClear) {
// Disconnect and/or ban depending on the node bloom services flag and // Disconnect and/or ban depending on the node bloom services flag and
// negotiated protocol version. // negotiated protocol version.
if !sp.enforceNodeBloomFlag(msg.Command()) { if !sp.enforceNodeBloomFlag(msg.Command()) {
@ -829,8 +833,8 @@ func (sp *serverPeer) OnFilterClear(p *peer.Peer, msg *wire.MsgFilterClear) {
if !sp.filter.IsLoaded() { if !sp.filter.IsLoaded() {
peerLog.Debugf("%s sent a filterclear request with no "+ peerLog.Debugf("%s sent a filterclear request with no "+
"filter loaded -- disconnecting", p) "filter loaded -- disconnecting", sp)
p.Disconnect() sp.Disconnect()
return return
} }
@ -842,7 +846,7 @@ func (sp *serverPeer) OnFilterClear(p *peer.Peer, msg *wire.MsgFilterClear) {
// delivering merkle blocks and associated transactions that match the filter. // delivering merkle blocks and associated transactions that match the filter.
// The peer will be disconnected if the server is not configured to allow bloom // The peer will be disconnected if the server is not configured to allow bloom
// filters. // filters.
func (sp *serverPeer) OnFilterLoad(p *peer.Peer, msg *wire.MsgFilterLoad) { func (sp *serverPeer) OnFilterLoad(_ *peer.Peer, msg *wire.MsgFilterLoad) {
// Disconnect and/or ban depending on the node bloom services flag and // Disconnect and/or ban depending on the node bloom services flag and
// negotiated protocol version. // negotiated protocol version.
if !sp.enforceNodeBloomFlag(msg.Command()) { if !sp.enforceNodeBloomFlag(msg.Command()) {
@ -857,7 +861,7 @@ func (sp *serverPeer) OnFilterLoad(p *peer.Peer, msg *wire.MsgFilterLoad) {
// OnGetAddr is invoked when a peer receives a getaddr bitcoin message // OnGetAddr is invoked when a peer receives a getaddr bitcoin message
// and is used to provide the peer with known addresses from the address // and is used to provide the peer with known addresses from the address
// manager. // manager.
func (sp *serverPeer) OnGetAddr(p *peer.Peer, msg *wire.MsgGetAddr) { func (sp *serverPeer) OnGetAddr(_ *peer.Peer, msg *wire.MsgGetAddr) {
// Don't return any addresses when running on the simulation test // Don't return any addresses when running on the simulation test
// network. This helps prevent the network from becoming another // network. This helps prevent the network from becoming another
// public test network since it will not be able to learn about other // public test network since it will not be able to learn about other
@ -868,7 +872,7 @@ func (sp *serverPeer) OnGetAddr(p *peer.Peer, msg *wire.MsgGetAddr) {
// Do not accept getaddr requests from outbound peers. This reduces // Do not accept getaddr requests from outbound peers. This reduces
// fingerprinting attacks. // fingerprinting attacks.
if !p.Inbound() { if !sp.Inbound() {
return return
} }
@ -881,7 +885,7 @@ func (sp *serverPeer) OnGetAddr(p *peer.Peer, msg *wire.MsgGetAddr) {
// OnAddr is invoked when a peer receives an addr bitcoin message and is // OnAddr is invoked when a peer receives an addr bitcoin message and is
// used to notify the server about advertised addresses. // used to notify the server about advertised addresses.
func (sp *serverPeer) OnAddr(p *peer.Peer, msg *wire.MsgAddr) { func (sp *serverPeer) OnAddr(_ *peer.Peer, msg *wire.MsgAddr) {
// Ignore addresses when running on the simulation test network. This // Ignore addresses when running on the simulation test network. This
// helps prevent the network from becoming another public test network // helps prevent the network from becoming another public test network
// since it will not be able to learn about other peers that have not // since it will not be able to learn about other peers that have not
@ -891,21 +895,21 @@ func (sp *serverPeer) OnAddr(p *peer.Peer, msg *wire.MsgAddr) {
} }
// Ignore old style addresses which don't include a timestamp. // Ignore old style addresses which don't include a timestamp.
if p.ProtocolVersion() < wire.NetAddressTimeVersion { if sp.ProtocolVersion() < wire.NetAddressTimeVersion {
return return
} }
// A message that has no addresses is invalid. // A message that has no addresses is invalid.
if len(msg.AddrList) == 0 { if len(msg.AddrList) == 0 {
peerLog.Errorf("Command [%s] from %s does not contain any addresses", peerLog.Errorf("Command [%s] from %s does not contain any addresses",
msg.Command(), p) msg.Command(), sp)
p.Disconnect() sp.Disconnect()
return return
} }
for _, na := range msg.AddrList { for _, na := range msg.AddrList {
// Don't add more address if we're disconnecting. // Don't add more address if we're disconnecting.
if !p.Connected() { if !sp.Connected() {
return return
} }
@ -926,18 +930,18 @@ func (sp *serverPeer) OnAddr(p *peer.Peer, msg *wire.MsgAddr) {
// addresses, and last seen updates. // addresses, and last seen updates.
// XXX bitcoind gives a 2 hour time penalty here, do we want to do the // XXX bitcoind gives a 2 hour time penalty here, do we want to do the
// same? // same?
sp.server.addrManager.AddAddresses(msg.AddrList, p.NA()) sp.server.addrManager.AddAddresses(msg.AddrList, sp.NA())
} }
// OnRead is invoked when a peer receives a message and it is used to update // OnRead is invoked when a peer receives a message and it is used to update
// the bytes received by the server. // the bytes received by the server.
func (sp *serverPeer) OnRead(p *peer.Peer, bytesRead int, msg wire.Message, err error) { func (sp *serverPeer) OnRead(_ *peer.Peer, bytesRead int, msg wire.Message, err error) {
sp.server.AddBytesReceived(uint64(bytesRead)) sp.server.AddBytesReceived(uint64(bytesRead))
} }
// OnWrite is invoked when a peer sends a message and it is used to update // OnWrite is invoked when a peer sends a message and it is used to update
// the bytes sent by the server. // the bytes sent by the server.
func (sp *serverPeer) OnWrite(p *peer.Peer, bytesWritten int, msg wire.Message, err error) { func (sp *serverPeer) OnWrite(_ *peer.Peer, bytesWritten int, msg wire.Message, err error) {
sp.server.AddBytesSent(uint64(bytesWritten)) sp.server.AddBytesSent(uint64(bytesWritten))
} }