Defer Snappy decompression of P2P messages until worker thread processing

CHANGELOG.md

@@ -59,6 +59,7 @@ are provided with different values, using input as per the execution-apis spec i
 - Fix addMod case with 256bit moduluses [#10001](https://github.com/besu-eth/besu/pull/10001)
 - Performance improvements on MOD variant instructions while converting from byte[] to longs [#9976](https://github.com/besu-eth/besu/pull/9976) 
 - Implement DIV and SDIV with long limbs [#9923](https://github.com/besu-eth/besu/pull/9923)
+- Defer Snappy decompression of inbound P2P messages from the Netty I/O thread to the worker thread, reducing memory held in the transaction worker queue to compressed size [#10048](https://github.com/besu-eth/besu/pull/10048)
 
 ## 26.2.0
 

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/EthProtocolManager.java

@@ -37,6 +37,7 @@
 import org.hyperledger.besu.ethereum.p2p.network.ProtocolManager;
 import org.hyperledger.besu.ethereum.p2p.rlpx.connections.PeerConnection;
 import org.hyperledger.besu.ethereum.p2p.rlpx.connections.PeerConnection.PeerNotConnected;
+import org.hyperledger.besu.ethereum.p2p.rlpx.framing.FramingException;
 import org.hyperledger.besu.ethereum.p2p.rlpx.wire.Capability;
 import org.hyperledger.besu.ethereum.p2p.rlpx.wire.Message;
 import org.hyperledger.besu.ethereum.p2p.rlpx.wire.MessageData;
@@ -298,12 +299,12 @@ public void processMessage(final Capability capability, final Message message) {
       return;
     }
 
-    // This will handle responses
-    ethPeers.dispatchMessage(ethPeer, ethMessage, getSupportedProtocol());
-
-    // This will handle requests
     Optional<MessageData> maybeResponseData = Optional.empty();
     try {
+      // This will handle responses
+      ethPeers.dispatchMessage(ethPeer, ethMessage, getSupportedProtocol());
+
+      // This will handle requests
       if (EthProtocol.requestIdCompatible(code)) {
         final Map.Entry<BigInteger, MessageData> requestIdAndEthMessage =
             ethMessage.getData().unwrapMessageData();
@@ -314,6 +315,17 @@ public void processMessage(final Capability capability, final Message message) {
       } else {
         maybeResponseData = ethMessages.dispatch(ethMessage, capability);
       }
+    } catch (final FramingException e) {
+      LOG.atDebug()
+          .setMessage(
+              "Failed to decompress message with code {} (BREACH_OF_PROTOCOL), disconnecting: {}, {}")
+          .addArgument(code)
+          .addArgument(ethPeer::toString)
+          .addArgument(e::toString)
+          .log();
+
+      ethPeer.disconnect(
+          DisconnectMessage.DisconnectReason.BREACH_OF_PROTOCOL_MALFORMED_MESSAGE_RECEIVED);
     } catch (final RLPException e) {
       LOG.atDebug()
           .setMessage(

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/snap/SnapProtocolManager.java

@@ -24,6 +24,7 @@
 import org.hyperledger.besu.ethereum.eth.sync.snapsync.SnapSyncConfiguration;
 import org.hyperledger.besu.ethereum.p2p.network.ProtocolManager;
 import org.hyperledger.besu.ethereum.p2p.rlpx.connections.PeerConnection;
+import org.hyperledger.besu.ethereum.p2p.rlpx.framing.FramingException;
 import org.hyperledger.besu.ethereum.p2p.rlpx.wire.AbstractSnapMessageData;
 import org.hyperledger.besu.ethereum.p2p.rlpx.wire.Capability;
 import org.hyperledger.besu.ethereum.p2p.rlpx.wire.Message;
@@ -113,18 +114,25 @@ public void processMessage(final Capability cap, final Message message) {
       return;
     }
 
-    // This will handle responses
-    ethPeers.dispatchMessage(ethPeer, ethMessage, getSupportedProtocol());
-
-    // This will handle requests
     Optional<MessageData> maybeResponseData = Optional.empty();
     try {
+      // This will handle responses
+      ethPeers.dispatchMessage(ethPeer, ethMessage, getSupportedProtocol());
+
+      // This will handle requests
       final Map.Entry<BigInteger, MessageData> requestIdAndEthMessage =
           ethMessage.getData().unwrapMessageData();
       maybeResponseData =
           snapMessages
               .dispatch(new EthMessage(ethPeer, requestIdAndEthMessage.getValue()), cap)
               .map(responseData -> responseData.wrapMessageData(requestIdAndEthMessage.getKey()));
+    } catch (final FramingException e) {
+      LOG.debug(
+          "Failed to decompress message with code {} (BREACH_OF_PROTOCOL), disconnecting: {}",
+          code,
+          ethPeer,
+          e);
+      ethPeer.disconnect(DisconnectReason.BREACH_OF_PROTOCOL_MALFORMED_MESSAGE_RECEIVED);
     } catch (final RLPException e) {
       LOG.debug(
           "Received malformed message code={} data={} (BREACH_OF_PROTOCOL), disconnecting: {}",

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/task/AbstractPeerRequestTask.java

@@ -20,6 +20,7 @@
 import org.hyperledger.besu.ethereum.eth.manager.PendingPeerRequest;
 import org.hyperledger.besu.ethereum.eth.manager.RequestManager;
 import org.hyperledger.besu.ethereum.eth.manager.exceptions.PeerBreachedProtocolException;
+import org.hyperledger.besu.ethereum.p2p.rlpx.framing.FramingException;
 import org.hyperledger.besu.ethereum.p2p.rlpx.wire.MessageData;
 import org.hyperledger.besu.ethereum.p2p.rlpx.wire.messages.DisconnectMessage.DisconnectReason;
 import org.hyperledger.besu.ethereum.rlp.RLPException;
@@ -111,6 +112,14 @@ private void handleMessage(
             promise.complete(r);
             peer.recordUsefulResponse();
           });
+    } catch (final FramingException e) {
+      // Peer sent us data that failed to decompress - disconnect
+      LOG.debug(
+          "Disconnecting with BREACH_OF_PROTOCOL due to decompression failure: {}",
+          peer.getLoggableId(),
+          e);
+      peer.disconnect(DisconnectReason.BREACH_OF_PROTOCOL_MALFORMED_MESSAGE_RECEIVED);
+      promise.completeExceptionally(new PeerBreachedProtocolException());
     } catch (final RLPException e) {
       // Peer sent us malformed data - disconnect
       LOG.debug(

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/NewPooledTransactionHashesMessageHandler.java

@@ -22,12 +22,19 @@
 import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.messages.NewPooledTransactionHashesMessage;
 import org.hyperledger.besu.ethereum.p2p.rlpx.wire.Capability;
+import org.hyperledger.besu.ethereum.p2p.rlpx.wire.MessageData;
+import org.hyperledger.besu.ethereum.p2p.rlpx.wire.messages.DisconnectMessage.DisconnectReason;
 
 import java.time.Duration;
 import java.time.Instant;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 class NewPooledTransactionHashesMessageHandler implements EthMessages.MessageCallback {
+  private static final Logger LOG =
+      LoggerFactory.getLogger(NewPooledTransactionHashesMessageHandler.class);
 
   private final NewPooledTransactionHashesMessageProcessor transactionsMessageProcessor;
   private final EthScheduler scheduler;
@@ -47,13 +54,30 @@ public NewPooledTransactionHashesMessageHandler(
   public void exec(final EthMessage message) {
     if (isEnabled.get()) {
       final Capability capability = message.getPeer().getConnection().capability(EthProtocol.NAME);
-      final NewPooledTransactionHashesMessage transactionsMessage =
-          NewPooledTransactionHashesMessage.readFrom(message.getData(), capability);
+      final MessageData rawMessage = message.getData();
       final Instant startedAt = now();
       scheduler.scheduleTxWorkerTask(
-          () ->
-              transactionsMessageProcessor.processNewPooledTransactionHashesMessage(
-                  message.getPeer(), transactionsMessage, startedAt, txMsgKeepAlive));
+          () -> {
+            if (message.getPeer().isDisconnected()) {
+              return;
+            }
+            final NewPooledTransactionHashesMessage transactionsMessage;
+            try {
+              transactionsMessage =
+                  NewPooledTransactionHashesMessage.readFrom(rawMessage, capability);
+            } catch (final Exception e) {
+              LOG.debug(
+                  "Malformed pooled transaction hashes message received (BREACH_OF_PROTOCOL), disconnecting: {}",
+                  message.getPeer(),
+                  e);
+              message
+                  .getPeer()
+                  .disconnect(DisconnectReason.BREACH_OF_PROTOCOL_MALFORMED_MESSAGE_RECEIVED);
+              return;
+            }
+            transactionsMessageProcessor.processNewPooledTransactionHashesMessage(
+                message.getPeer(), transactionsMessage, startedAt, txMsgKeepAlive);
+          });
     }
   }
 

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/TransactionsMessageHandler.java

@@ -20,12 +20,18 @@
 import org.hyperledger.besu.ethereum.eth.manager.EthMessages;
 import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.messages.TransactionsMessage;
+import org.hyperledger.besu.ethereum.p2p.rlpx.wire.MessageData;
+import org.hyperledger.besu.ethereum.p2p.rlpx.wire.messages.DisconnectMessage.DisconnectReason;
 
 import java.time.Duration;
 import java.time.Instant;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 class TransactionsMessageHandler implements EthMessages.MessageCallback {
+  private static final Logger LOG = LoggerFactory.getLogger(TransactionsMessageHandler.class);
 
   private final TransactionsMessageProcessor transactionsMessageProcessor;
   private final EthScheduler scheduler;
@@ -44,13 +50,29 @@ public TransactionsMessageHandler(
   @Override
   public void exec(final EthMessage message) {
     if (isEnabled.get()) {
-      final TransactionsMessage transactionsMessage =
-          TransactionsMessage.readFrom(message.getData());
+      final MessageData rawMessage = message.getData();
       final Instant startedAt = now();
       scheduler.scheduleTxWorkerTask(
-          () ->
-              transactionsMessageProcessor.processTransactionsMessage(
-                  message.getPeer(), transactionsMessage, startedAt, txMsgKeepAlive));
+          () -> {
+            if (message.getPeer().isDisconnected()) {
+              return;
+            }
+            final TransactionsMessage transactionsMessage;
+            try {
+              transactionsMessage = TransactionsMessage.readFrom(rawMessage);
+            } catch (final Exception e) {
+              LOG.debug(
+                  "Malformed transactions message received (BREACH_OF_PROTOCOL), disconnecting: {}",
+                  message.getPeer(),
+                  e);
+              message
+                  .getPeer()
+                  .disconnect(DisconnectReason.BREACH_OF_PROTOCOL_MALFORMED_MESSAGE_RECEIVED);
+              return;
+            }
+            transactionsMessageProcessor.processTransactionsMessage(
+                message.getPeer(), transactionsMessage, startedAt, txMsgKeepAlive);
+          });
     }
   }
 

ethereum/p2p/src/main/java/org/hyperledger/besu/ethereum/p2p/rlpx/connections/netty/ApiHandler.java

@@ -99,15 +99,17 @@ protected void channelRead0(final ChannelHandlerContext ctx, final MessageData o
       }
       return;
     }
-    LOG.atTrace()
-        .addMarker(P2P_MESSAGE_MARKER)
-        .setMessage("Received {} from {} via protocol {}")
-        .addArgument(message)
-        .addArgument(connection.getPeerInfo())
-        .addArgument(demultiplexed.getCapability())
-        .addKeyValue("rawData", message.getData())
-        .addKeyValue("decodedData", message::toStringDecoded)
-        .log();
+    if (LOG.isTraceEnabled()) {
+      LOG.atTrace()
+          .addMarker(P2P_MESSAGE_MARKER)
+          .setMessage("Received {} from {} via protocol {}")
+          .addArgument(message)
+          .addArgument(connection.getPeerInfo())
+          .addArgument(demultiplexed.getCapability())
+          .addKeyValue("rawData", message.getData())
+          .addKeyValue("decodedData", message::toStringDecoded)
+          .log();
+    }
 
     connectionEventDispatcher.dispatchMessage(demultiplexed.getCapability(), connection, message);
   }

ethereum/p2p/src/main/java/org/hyperledger/besu/ethereum/p2p/rlpx/connections/netty/DeFramer.java

@@ -31,6 +31,7 @@
 import org.hyperledger.besu.ethereum.p2p.rlpx.wire.CapabilityMultiplexer;
 import org.hyperledger.besu.ethereum.p2p.rlpx.wire.MessageData;
 import org.hyperledger.besu.ethereum.p2p.rlpx.wire.PeerInfo;
+import org.hyperledger.besu.ethereum.p2p.rlpx.wire.RawMessage;
 import org.hyperledger.besu.ethereum.p2p.rlpx.wire.SubProtocol;
 import org.hyperledger.besu.ethereum.p2p.rlpx.wire.messages.DisconnectMessage;
 import org.hyperledger.besu.ethereum.p2p.rlpx.wire.messages.HelloMessage;
@@ -54,6 +55,7 @@
 import io.netty.handler.codec.ByteToMessageDecoder;
 import io.netty.handler.codec.DecoderException;
 import io.netty.handler.timeout.IdleStateHandler;
+import org.apache.tuweni.bytes.Bytes;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -227,7 +229,9 @@ protected void decode(final ChannelHandlerContext ctx, final ByteBuf in, final L
             "Message received before HELLO's exchanged (BREACH_OF_PROTOCOL), disconnecting.  Peer: {}, Code: {}, Data: {}",
             expectedPeer.map(Peer::getEnodeURLString).orElse("unknown"),
             message.getCode(),
-            message.getData().toString());
+            message instanceof RawMessage raw && raw.getCompressedData() != null
+                ? "snappy compressed data: " + Bytes.wrap(raw.getCompressedData())
+                : message.getData().toString());
         ctx.writeAndFlush(
                 new OutboundMessage(
                     null,

ethereum/p2p/src/main/java/org/hyperledger/besu/ethereum/p2p/rlpx/framing/Framer.java

@@ -282,38 +282,35 @@ private MessageData processFrame(final ByteBuf f, final int frameSize) {
     final int id = idbv.isZero() || idbv.size() == 0 ? 0 : idbv.get(0);
 
     // Write message data to ByteBuf, decompressing as necessary
-    final Bytes data;
     if (compressionEnabled) {
       final byte[] compressedMessageData = Arrays.copyOfRange(frameData, 1, frameData.length - pad);
       final int uncompressedLength = compressor.uncompressedLength(compressedMessageData);
       if (uncompressedLength >= LENGTH_MAX_MESSAGE_FRAME) {
         throw error("Message size %s in excess of maximum length.", uncompressedLength);
       }
-      Bytes _data;
-      try {
-        final byte[] decompressedMessageData = compressor.decompress(compressedMessageData);
-        _data = Bytes.wrap(decompressedMessageData);
-        compressionSuccessful = true;
-      } catch (final FramingException fe) {
-        if (compressionSuccessful) {
-          throw fe;
-        } else {
-          // OpenEthereum/Parity does not implement EIP-706
-          // If failing on the first packet downgrade to uncompressed
+
+      if (!compressionSuccessful) {
+        // First compressed message: decompress eagerly to validate and handle
+        // the OpenEthereum/Parity fallback (non-Snappy peer detection via EIP-706)
+        try {
+          final byte[] decompressedMessageData = compressor.decompress(compressedMessageData);
+          compressionSuccessful = true;
+          return new RawMessage(id, Bytes.wrap(decompressedMessageData));
+        } catch (final FramingException fe) {
           compressionEnabled = false;
           LOG.debug("Snappy decompression failed: downgrading to uncompressed");
           final int messageLength = frameSize - LENGTH_MESSAGE_ID;
-          _data = Bytes.wrap(frameData, 1, messageLength);
+          return new RawMessage(id, Bytes.wrap(frameData, 1, messageLength));
         }
+      } else {
+        // Subsequent messages: store compressed, decompress lazily on getData()
+        return new RawMessage(id, compressedMessageData);
       }
-      data = _data;
     } else {
-      // Move data to a ByteBuf
       final int messageLength = frameSize - LENGTH_MESSAGE_ID;
-      data = Bytes.wrap(frameData, 1, messageLength);
+      final Bytes data = Bytes.wrap(frameData, 1, messageLength);
+      return new RawMessage(id, data);
     }
-
-    return new RawMessage(id, data);
   }
 
   private void validateMac(final byte[] candidateMac, final byte[] expectedMac) {

ethereum/p2p/src/main/java/org/hyperledger/besu/ethereum/p2p/rlpx/wire/AbstractMessageData.java

@@ -27,7 +27,7 @@ protected AbstractMessageData(final Bytes data) {
   }
 
   @Override
-  public final int getSize() {
+  public int getSize() {
     return data.size();
   }