fix: prevent pipeline errors from killing the Reactor subscription

The pipeline used doOnNext to chain handlers, but onErrorContinue does
not work reliably with doOnNext (a known Reactor limitation). When a
StackOverflowError escaped a handler, it could intermittently terminate
the Reactor subscription, causing the node to stop processing all
subsequent incoming packets.

Wrap each handler invocation in a try-catch within the pipeline so errors
are contained at the handler level and never propagate to Reactor.

Also add catch(Throwable) to HandshakeMessagePacketHandler for
defense-in-depth, matching the existing pattern in MessagePacketHandler.

Author: usmansaleem

src/main/java/org/ethereum/beacon/discovery/pipeline/PipelineImpl.java

@@ -29,7 +29,15 @@ public class PipelineImpl implements Pipeline {
   public synchronized Pipeline build() {
     started.set(true);
     for (EnvelopeHandler handler : envelopeHandlers) {
-      pipeline = pipeline.doOnNext(handler::handle);
+      pipeline =
+          pipeline.doOnNext(
+              envelope -> {
+                try {
+                  handler.handle(envelope);
+                } catch (Throwable t) {
+                  LOG.debug("Error in pipeline handler {}: {}", handler.getClass().getSimpleName(), t.getMessage());
+                }
+              });
     }
     Flux.from(pipeline)
         .onErrorContinue(

src/main/java/org/ethereum/beacon/discovery/pipeline/handler/HandshakeMessagePacketHandler.java

@@ -151,6 +151,13 @@ public void handle(Envelope envelope) {
               packet, session.getNodeRecord(), session.getState()),
           ex);
       markHandshakeAsFailed(envelope, session);
+    } catch (Throwable t) {
+      LOG.warn(
+          "Unexpected error while processing handshake [{}] from node {}",
+          packet,
+          session.getNodeRecord(),
+          t);
+      markHandshakeAsFailed(envelope, session);
     }
   }