From 696fee93ec5bc5585057942f52bdb56f19521d20 Mon Sep 17 00:00:00 2001 From: Giovanni Gargiulo Date: Wed, 24 Jun 2026 23:55:20 +0100 Subject: [PATCH 1/3] fix(txsubmission): handle client MsgDone in server agent to avoid mempool stall The N2N TxSubmission server agent never handled a client MsgDone. When a peer terminated the protocol with MsgDone while the server was blocking-waiting in TxIdsBlocking, TxSubmissionState.TxIdsBlocking.nextState() returned `this`, so the agent stayed wedged in TxIdsBlocking forever. Because the client holds agency in that state, the server could never send another RequestTxIds ("Cannot request tx IDs in state: TxIdsBlocking"), and since the agent never became Done, MiniProtoServerInboundHandler kept it registered ("No agent found to handle protocol ..."). Net effect: the node silently stops ingesting mempool transactions while chain-sync/block-fetch keep working. Fix: transition to Done on MsgDone from the client-agency states (TxIdsBlocking per spec; TxIdsNonBlocking/Txs defensively), and handle MsgDone in TxSubmissionServerAgent.processResponse so it terminates cleanly instead of logging "Unexpected message type". Once Done, the inbound handler stops routing to the agent and a fresh agent is created when the peer reconnects. Adds unit tests for the MsgDone -> Done transition. --- .../txsubmission/TxSubmissionServerAgent.java | 14 ++++++++++ .../txsubmission/TxSubmissionState.java | 12 +++++++++ .../TxSubmissionServerAgentTest.java | 27 +++++++++++++++++++ 3 files changed, 53 insertions(+) diff --git a/core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionServerAgent.java b/core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionServerAgent.java index b3098a88..c6869def 100644 --- a/core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionServerAgent.java +++ b/core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionServerAgent.java @@ -114,11 +114,25 @@ public void processResponse(Message message) { handleReplyTxIds((ReplyTxIds) message); } else if (message instanceof ReplyTxs) { handleReplyTxs((ReplyTxs) message); + } else if (message instanceof MsgDone) { + handleDone(); } else { log.warn("Unexpected message type: {}", message.getClass().getSimpleName()); } } + /** + * The client terminated the tx-submission protocol with MsgDone. The state machine has already + * advanced to {@link TxSubmissionState#Done} (see {@link TxSubmissionState}), so the inbound + * handler stops routing to this agent and the session can be torn down; a fresh agent is created + * when the peer reconnects. Drop any in-flight bookkeeping so nothing leaks. + */ + private void handleDone() { + log.info("Received MsgDone from client - tx-submission protocol terminated, agent is Done"); + outstandingTxIds.clear(); + pendingRequest = null; + } + private void handleInit() { log.info("Received Init message from client - transitioning to Idle state"); diff --git a/core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionState.java b/core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionState.java index fab5bd1a..ef0e5175 100644 --- a/core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionState.java +++ b/core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionState.java @@ -2,6 +2,7 @@ import com.bloxbean.cardano.yaci.core.protocol.Message; import com.bloxbean.cardano.yaci.core.protocol.State; +import com.bloxbean.cardano.yaci.core.protocol.txsubmission.messges.MsgDone; import com.bloxbean.cardano.yaci.core.protocol.txsubmission.messges.ReplyTxIds; import com.bloxbean.cardano.yaci.core.protocol.txsubmission.messges.ReplyTxs; import com.bloxbean.cardano.yaci.core.protocol.txsubmission.messges.RequestTxIds; @@ -42,8 +43,15 @@ public boolean hasAgency(boolean isClient) { TxIdsBlocking { @Override public State nextState(Message message) { + // Per the node-to-node tx-submission mini-protocol, the client may terminate the + // protocol with MsgDone instead of ReplyTxIds while the server is blocking-waiting. + // Transition to Done so the agent can be torn down and re-created on reconnect; + // without this the server agent stays wedged here forever (client holds agency, so + // it can never send another RequestTxIds) and stops ingesting mempool txs. if (message instanceof ReplyTxIds) return Idle; + else if (message instanceof MsgDone) + return Done; else return this; } @@ -58,6 +66,8 @@ public boolean hasAgency(boolean isClient) { public State nextState(Message message) { if (message instanceof ReplyTxIds) return Idle; + else if (message instanceof MsgDone) // defensive: honour client termination + return Done; else return this; } @@ -72,6 +82,8 @@ public boolean hasAgency(boolean isClient) { public State nextState(Message message) { if (message instanceof ReplyTxs) return Idle; + else if (message instanceof MsgDone) // defensive: honour client termination + return Done; else return this; } diff --git a/core/src/test/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionServerAgentTest.java b/core/src/test/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionServerAgentTest.java index 445a4a10..435e56ba 100644 --- a/core/src/test/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionServerAgentTest.java +++ b/core/src/test/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionServerAgentTest.java @@ -332,6 +332,33 @@ void testEmptyReplyTxIds() { assertEquals(0, requestTxIds.getAckTxIds()); // No transactions to acknowledge } + @Test + void testMsgDoneFromTxIdsBlockingTransitionsToDone() { + // Move to Idle, then have the server send a blocking RequestTxIds so we land in TxIdsBlocking + agent.receiveResponse(new Init()); + assertEquals(TxSubmissionState.Idle, agent.getCurrentState()); + + agent.sendRequest(new RequestTxIds(true, (short) 0, (short) 10)); + assertEquals(TxSubmissionState.TxIdsBlocking, agent.getCurrentState()); + assertFalse(agent.isDone()); + + // Client terminates the protocol with MsgDone while we are blocking-waiting. + // Before the fix the agent stayed wedged in TxIdsBlocking forever; now it must reach Done. + agent.receiveResponse(new MsgDone()); + + assertEquals(TxSubmissionState.Done, agent.getCurrentState()); + assertTrue(agent.isDone()); + assertEquals(0, agent.getOutstandingTxCount()); + } + + @Test + void testMsgDoneStateTransitionsForAllClientAgencyStates() { + // The terminal MsgDone transition must hold for every state where the client has agency. + assertEquals(TxSubmissionState.Done, TxSubmissionState.TxIdsBlocking.nextState(new MsgDone())); + assertEquals(TxSubmissionState.Done, TxSubmissionState.TxIdsNonBlocking.nextState(new MsgDone())); + assertEquals(TxSubmissionState.Done, TxSubmissionState.Txs.nextState(new MsgDone())); + } + @Test void testConfigurationValidation() { // Valid config should not log warnings From 11fea20f5e1479bd834ba4105e961ef15c4f7c6f Mon Sep 17 00:00:00 2001 From: Satya <35016438+satran004@users.noreply.github.com> Date: Mon, 29 Jun 2026 15:04:59 +0800 Subject: [PATCH 2/3] Remove comments from TxIdsBlocking state logic Removed comments explaining state transition logic in TxIdsBlocking. --- .../yaci/core/protocol/txsubmission/TxSubmissionState.java | 5 ----- 1 file changed, 5 deletions(-) diff --git a/core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionState.java b/core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionState.java index ef0e5175..6e3b6da1 100644 --- a/core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionState.java +++ b/core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionState.java @@ -43,11 +43,6 @@ public boolean hasAgency(boolean isClient) { TxIdsBlocking { @Override public State nextState(Message message) { - // Per the node-to-node tx-submission mini-protocol, the client may terminate the - // protocol with MsgDone instead of ReplyTxIds while the server is blocking-waiting. - // Transition to Done so the agent can be torn down and re-created on reconnect; - // without this the server agent stays wedged here forever (client holds agency, so - // it can never send another RequestTxIds) and stops ingesting mempool txs. if (message instanceof ReplyTxIds) return Idle; else if (message instanceof MsgDone) From 53eea242ce4d78b3eb454d2153da3c882973738d Mon Sep 17 00:00:00 2001 From: Giovanni Gargiulo Date: Sat, 18 Jul 2026 23:08:15 +0100 Subject: [PATCH 3/3] fix(txsubmission): drop non-spec MsgDone transitions, keep only TxIdsBlocking Per review feedback, MsgDone is only legal in TxIdsBlocking under the node-to-node tx-submission spec; the client can never send it in TxIdsNonBlocking or Txs. Remove the two defensive transitions (which also tripped the SonarQube duplication gate) and update the test to assert those states ignore MsgDone. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../core/protocol/txsubmission/TxSubmissionState.java | 4 ---- .../txsubmission/TxSubmissionServerAgentTest.java | 10 ++++++---- 2 files changed, 6 insertions(+), 8 deletions(-) diff --git a/core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionState.java b/core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionState.java index 6e3b6da1..4ca74521 100644 --- a/core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionState.java +++ b/core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionState.java @@ -61,8 +61,6 @@ public boolean hasAgency(boolean isClient) { public State nextState(Message message) { if (message instanceof ReplyTxIds) return Idle; - else if (message instanceof MsgDone) // defensive: honour client termination - return Done; else return this; } @@ -77,8 +75,6 @@ public boolean hasAgency(boolean isClient) { public State nextState(Message message) { if (message instanceof ReplyTxs) return Idle; - else if (message instanceof MsgDone) // defensive: honour client termination - return Done; else return this; } diff --git a/core/src/test/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionServerAgentTest.java b/core/src/test/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionServerAgentTest.java index 435e56ba..027f10ab 100644 --- a/core/src/test/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionServerAgentTest.java +++ b/core/src/test/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionServerAgentTest.java @@ -352,11 +352,13 @@ void testMsgDoneFromTxIdsBlockingTransitionsToDone() { } @Test - void testMsgDoneStateTransitionsForAllClientAgencyStates() { - // The terminal MsgDone transition must hold for every state where the client has agency. + void testMsgDoneStateTransition() { + // Per the node-to-node tx-submission spec, MsgDone is only legal (and only handled) + // in TxIdsBlocking. The other client-agency states are not part of the spec for MsgDone, + // so they must ignore it and stay put rather than transition to Done. assertEquals(TxSubmissionState.Done, TxSubmissionState.TxIdsBlocking.nextState(new MsgDone())); - assertEquals(TxSubmissionState.Done, TxSubmissionState.TxIdsNonBlocking.nextState(new MsgDone())); - assertEquals(TxSubmissionState.Done, TxSubmissionState.Txs.nextState(new MsgDone())); + assertEquals(TxSubmissionState.TxIdsNonBlocking, TxSubmissionState.TxIdsNonBlocking.nextState(new MsgDone())); + assertEquals(TxSubmissionState.Txs, TxSubmissionState.Txs.nextState(new MsgDone())); } @Test