network:ws block byte limiter

config/localTemplate.go

@@ -521,9 +521,9 @@ type Local struct {
 	// only relevant if TxIncomingFilteringFlags is non-zero
 	TxIncomingFilterMaxSize uint64 `version[28]:"500000"`
 
-	// BlockServiceHTTPMemCap is the memory capacity in bytes which is allowed for the block service to use for HTTP block requests.
+	// BlockServiceMemCap is the memory capacity in bytes which is allowed for the block service to use for HTTP block requests.
 	// When it exceeds this capacity, it redirects the block requests to a different node
-	BlockServiceHTTPMemCap uint64 `version[28]:"500000000"`
+	BlockServiceMemCap uint64 `version[28]:"500000000"`
 }
 
 // DNSBootstrapArray returns an array of one or more DNS Bootstrap identifiers

config/local_defaults.go

@@ -30,7 +30,7 @@ var defaultLocal = Local{
 	Archival:                                   false,
 	BaseLoggerDebugLevel:                       4,
 	BlockServiceCustomFallbackEndpoints:        "",
-	BlockServiceHTTPMemCap:                     500000000,
+	BlockServiceMemCap:                         500000000,
 	BroadcastConnectionsLimit:                  -1,
 	CadaverDirectory:                           "",
 	CadaverSizeTarget:                          0,

installer/config.json.example

@@ -9,7 +9,7 @@
     "Archival": false,
     "BaseLoggerDebugLevel": 4,
     "BlockServiceCustomFallbackEndpoints": "",
-    "BlockServiceHTTPMemCap": 500000000,
+    "BlockServiceMemCap": 500000000,
     "BroadcastConnectionsLimit": -1,
     "CadaverDirectory": "",
     "CadaverSizeTarget": 0,

network/wsNetwork.go

@@ -142,6 +142,8 @@
 var networkPrioPPCompressedSize = metrics.MakeCounter(metrics.MetricName{Name: "algod_network_prio_pp_compressed_size_total", Description: "cumulative size of all compressed PP"})
 var networkPrioPPNonCompressedSize = metrics.MakeCounter(metrics.MetricName{Name: "algod_network_prio_pp_non_compressed_size_total", Description: "cumulative size of all non-compressed PP"})
 
+var networkCatchupMessagesDropped = metrics.MakeCounter(metrics.MetricName{Name: "algod_network_ue_messages_dropped", Description: "number of (UE) block catchup request messages dropped due to being at byte limit"})
+
 // peerDisconnectionAckDuration defines the time we would wait for the peer disconnection to complete.
 const peerDisconnectionAckDuration = 5 * time.Second
 
@@ -250,6 +252,8 @@
 	// is used to ensure fairness across peers in terms of processing
 	// messages.
 	processing chan struct{}
+
+	Callback func()
 }
 
 // Tag is a short string (2 bytes) marking a type of message

network/wsNetwork_test.go

@@ -605,7 +605,7 @@ func TestWebsocketNetworkCancel(t *testing.T) {
 	msgs[50].ctx = ctx
 
 	for _, peer := range peers {
-		peer.sendBufferHighPrio <- sendMessages{msgs}
+		peer.sendBufferHighPrio <- sendMessages{msgs: msgs}
 	}
 
 	select {

network/wsPeer.go

@@ -173,7 +173,8 @@ type Response struct {
 }
 
 type sendMessages struct {
-	msgs []sendMessage
+	msgs     []sendMessage
+	callback func()
 }
 
 type wsPeer struct {
@@ -410,7 +411,7 @@ func (wp *wsPeer) Respond(ctx context.Context, reqMsg IncomingMessage, responseT
 	}
 
 	select {
-	case wp.sendBufferBulk <- sendMessages{msgs: msg}:
+	case wp.sendBufferBulk <- sendMessages{msgs: msg, callback: reqMsg.Callback}:
 	case <-wp.closing:
 		wp.net.log.Debugf("peer closing %s", wp.conn.RemoteAddr().String())
 		return
@@ -713,6 +714,9 @@ func (wp *wsPeer) handleFilterMessage(msg IncomingMessage) {
 }
 
 func (wp *wsPeer) writeLoopSend(msgs sendMessages) disconnectReason {
+	if msgs.callback != nil {
+		defer msgs.callback()
+	}
 	for _, msg := range msgs.msgs {
 		select {
 		case <-msg.ctx.Done():
@@ -730,17 +734,17 @@ func (wp *wsPeer) writeLoopSend(msgs sendMessages) disconnectReason {
 }
 
 func (wp *wsPeer) writeLoopSendMsg(msg sendMessage) disconnectReason {
-	if len(msg.data) > maxMessageLength {
-		wp.net.log.Errorf("trying to send a message longer than we would receive: %d > %d tag=%s", len(msg.data), maxMessageLength, string(msg.data[0:2]))
-		// just drop it, don't break the connection
-		return disconnectReasonNone
-	}
 	if msg.msgTags != nil {
 		// when msg.msgTags is non-nil, the read loop has received a message-of-interest message that we want to apply.
 		// in order to avoid any locking, it sent it to this queue so that we could set it as the new outgoing message tag filter.
 		wp.sendMessageTag = msg.msgTags
 		return disconnectReasonNone
 	}
+	if len(msg.data) > maxMessageLength {
+		wp.net.log.Errorf("trying to send a message longer than we would receive: %d > %d tag=%s", len(msg.data), maxMessageLength, string(msg.data[0:2]))
+		// just drop it, don't break the connection
+		return disconnectReasonNone
+	}
 	// the tags are always 2 char long; note that this is safe since it's only being used for messages that we have generated locally.
 	tag := protocol.Tag(msg.data[:2])
 	if !wp.sendMessageTag[tag] {
@@ -804,9 +808,15 @@ func (wp *wsPeer) writeLoop() {
 			}
 		case data := <-wp.sendBufferBulk:
 			if writeErr := wp.writeLoopSend(data); writeErr != disconnectReasonNone {
+				if data.callback != nil {
+					data.callback()
+				}
 				cleanupCloseError = writeErr
 				return
 			}
+			if data.callback != nil {
+				data.callback()
+			}
 		}
 	}
 }
@@ -920,6 +930,21 @@ func (wp *wsPeer) Close(deadline time.Time) {
 			wp.net.log.Infof("failed to CloseWithoutFlush to connection for %s", wp.conn.RemoteAddr().String())
 		}
 	}
+
+	// We need to loop through all of the messages with callbacks still in the send queue and call them
+	// to ensure that state of counters such as wsBlockBytesUsed is correct.
+L:
+	for {
+		select {
+		case msgs := <-wp.sendBufferBulk:
+			if msgs.callback != nil {
+				msgs.callback()
+			}
+		default:
+			break L
+		}
+
+	}
 	// now call all registered closers
 	for _, f := range wp.closers {
 		f()

rpcs/blockService.go

@@ -130,7 +130,7 @@ func MakeBlockService(log logging.Logger, config config.Local, ledger LedgerForB
 		fallbackEndpoints:       makeFallbackEndpoints(log, config.BlockServiceCustomFallbackEndpoints),
 		enableArchiverFallback:  config.EnableBlockServiceFallbackToArchiver,
 		log:                     log,
-		memoryCap:               config.BlockServiceHTTPMemCap,
+		memoryCap:               config.BlockServiceMemCap,
 	}
 	if service.enableService {
 		net.RegisterHTTPHandler(BlockServiceBlockPath, service)
@@ -301,8 +301,19 @@ const datatypeUnsupportedErrMsg = "requested data type is unsupported"
 func (bs *BlockService) handleCatchupReq(ctx context.Context, reqMsg network.IncomingMessage) {
 	target := reqMsg.Sender.(network.UnicastPeer)
 	var respTopics network.Topics
+	var n uint64
 
 	defer func() {
+		if n > 0 {
+			reqMsg.Callback = func() {
+				bs.mu.Lock()
+				bs.memoryUsed -= n
+				bs.mu.Unlock()
+			}
+			bs.mu.Lock()
+			bs.memoryUsed += n
+			bs.mu.Unlock()
+		}
 		target.Respond(ctx, reqMsg, respTopics)
 	}()
 
@@ -338,7 +349,7 @@ func (bs *BlockService) handleCatchupReq(ctx context.Context, reqMsg network.Inc
 				[]byte(roundNumberParseErrMsg))}
 		return
 	}
-	respTopics = topicBlockBytes(bs.log, bs.ledger, basics.Round(round), string(requestType))
+	respTopics, n = topicBlockBytes(bs.log, bs.ledger, basics.Round(round), string(requestType))
 	return
 }
 
@@ -416,7 +427,7 @@ func (bs *BlockService) rawBlockBytes(round basics.Round) ([]byte, error) {
 	return data, err
 }
 
-func topicBlockBytes(log logging.Logger, dataLedger LedgerForBlockService, round basics.Round, requestType string) network.Topics {
+func topicBlockBytes(log logging.Logger, dataLedger LedgerForBlockService, round basics.Round, requestType string) (network.Topics, uint64) {
 	blk, cert, err := dataLedger.EncodedBlockCert(round)
 	if err != nil {
 		switch err.(type) {
@@ -425,7 +436,7 @@ func topicBlockBytes(log logging.Logger, dataLedger LedgerForBlockService, round
 			log.Infof("BlockService topicBlockBytes: %s", err)
 		}
 		return network.Topics{
-			network.MakeTopic(network.ErrorKey, []byte(blockNotAvailableErrMsg))}
+			network.MakeTopic(network.ErrorKey, []byte(blockNotAvailableErrMsg))}, 0
 	}
 	switch requestType {
 	case BlockAndCertValue:
@@ -434,10 +445,10 @@ func topicBlockBytes(log logging.Logger, dataLedger LedgerForBlockService, round
 				BlockDataKey, blk),
 			network.MakeTopic(
 				CertDataKey, cert),
-		}
+		}, uint64(len(blk) + len(cert))
 	default:
 		return network.Topics{
-			network.MakeTopic(network.ErrorKey, []byte(datatypeUnsupportedErrMsg))}
+			network.MakeTopic(network.ErrorKey, []byte(datatypeUnsupportedErrMsg))}, 0
 	}
 }
 

test/testdata/configs/config-v28.json

@@ -9,7 +9,7 @@
     "Archival": false,
     "BaseLoggerDebugLevel": 4,
     "BlockServiceCustomFallbackEndpoints": "",
-    "BlockServiceHTTPMemCap": 500000000,
+    "BlockServiceMemCap": 500000000,
     "BroadcastConnectionsLimit": -1,
     "CadaverDirectory": "",
     "CadaverSizeTarget": 0,