Skip to content

fix(net): release sync requests when block deserialization fails #7093

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
317787106 wants to merge 6 commits into
base: release_v4.8.3
Choose a base branch
from
+250 −4
Open

fix(net): release sync requests when block deserialization fails #7093

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
6bd88e1
fix(net): release sync requests when block deserialization fails
317787106 Nov 6, 2026
f12c112
fix(net): reject unmatched end-group tags at block ingress
317787106 Nov 9, 2026
cfdbfcf
Merge branch 'release_v4.8.3' into hotfix/sync_block_deserialization
317787106 Nov 9, 2026
dee5f44
test(net): use valid witness addresses in sync regressions
317787106 Nov 9, 2026
7b1150b
fix(net): unify deferred sync block validation
317787106 Nov 10, 2026
31d366f
fix testcase BlockFetchFailureTest
317787106 Nov 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion framework/src/main/java/org/tron/core/net/message/adv/BlockMessage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
package org.tron.core.net.message.adv;

import com.google.protobuf.CodedInputStream;
import org.tron.common.overlay.message.Message;
import org.tron.common.utils.Sha256Hash;
import org.tron.core.capsule.BlockCapsule;
Expand All @@ -15,7 +16,10 @@ public class BlockMessage extends TronMessage {
public BlockMessage(byte[] data) throws Exception {
super(data);
this.type = MessageTypes.BLOCK.asByte();
this.block = new BlockCapsule(getCodedInputStream(data));
CodedInputStream input = getCodedInputStream(data);
this.block = new BlockCapsule(input);
// A top-level block must end at EOF, not at an unmatched END_GROUP tag.
input.checkLastTagWas(0);
if (Message.isFilter()) {
Message.compareBytes(data, block.getInstance().toByteArray());
TransactionCapsule.validContractProto(block.getInstance().getTransactionsList());
Expand Down
9 changes: 7 additions & 2 deletions framework/src/main/java/org/tron/core/net/service/sync/SyncService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -321,9 +321,14 @@ private synchronized void handleSyncBlock() {
isProcessed[0] = true;
BlockCapsule block;
try {
block = new BlockCapsule(unparsedBlock.getData());
block = new BlockMessage(unparsedBlock.getData()).getBlockCapsule();
} catch (Exception e) {

@bladehan1 bladehan1 Nov 6, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[NIT] Document the two-parse-entry asymmetry and proto-filter precondition at this catch site

This failure path is only reachable because the network-side parse (BlockMessage(byte[]) -> Block.parseFrom(CodedInputStream)) accepts a trailing 0x0c (END_GROUP) byte while the deferred BlockCapsule(byte[]) parse (Block.parseFrom(byte[])) rejects it — verified against protobuf-java 3.25.8. With proto filtering enabled (allowProtoFilterNum == 1), the entry path's Message.compareBytes length check rejects the payload first, so this catch is nearly unreachable there; on filter-disabled networks it is reachable. If a future protobuf upgrade unifies the two parse entries, or a maintainer 'unifies' them by hand, the reachability of this path changes silently and the regression test degrades into testing dead code.

Suggestion: add a short comment above this catch block noting that the second parse is stricter than the entry parse and that a failure here means the supplying peer is not trustworthy; re-verify the regression test semantics whenever protobuf-java is upgraded.

@317787106 317787106 Nov 9, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Updated in f12c112. BlockMessage now calls checkLastTagWas(0) after the initial parse, rejecting unmatched top-level END_GROUP tags regardless of the proto-filter setting. A comment at that check explains the EOF requirement.

The deferred-failure regression now explicitly injects damaged queued data to verify defensive cleanup independently of ingress validation. Separate tests cover ingress rejection and legitimate 0x0c bytes inside fields.

logger.warn("Deserialize block {} failed", blockId.getString(), e);
logger.warn("Deserialize block {} from {} failed",
blockId.getString(), peerConnection.getInetSocketAddress(), e);
// Deferred validation may fail after ingress, e.g. on ALLOW_PROTO_FILTER_NUM changes
// or interruption. Release sync state so this block remains eligible for retry.
peerConnection.getSyncBlockInProcess().remove(blockId);

@xxo1shine xxo1shine Nov 6, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[SHOULD] Please use the same deserialization path as the initial network parse:

block = new BlockMessage(unparsedBlock.getData()).getBlockCapsule();

The reproduced stall comes from using different Protobuf overloads: BlockMessage(byte[]) accepts the trailing 0x0c when filtering is disabled, while the deferred BlockCapsule(byte[]) rejects it. Reusing BlockMessage makes the two parses consistent and preserves its existing filtering and contract-format checks. This directly fixes the reproduced regression without adding a BAD_BLOCK disconnect for a representation already accepted at ingress.

Please adjust the regression to verify that, with filtering disabled, the block is processed successfully and the provider stays connected. With filtering enabled, the payload should still be rejected at the initial parse. Keep the existing catch; cleanup for other deferred-parsing failures can be considered separately.

@317787106 317787106 Nov 9, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I disagree with making both parsing paths accept this payload. A block message with an extra, unmatched trailing END_GROUP (0x0c) is malformed and must be rejected. The earlier ingress acceptance was caused by missing end-tag validation and should be fixed.

In f12c112, BlockMessage now calls checkLastTagWas(0) after the initial parse, rejecting this payload under both filter settings before the sync-state transition. Existing filtering and contract-format checks remain unchanged.

The tests verify ingress rejection, disconnect cleanup, and retry through another peer. Deferred-failure cleanup remains as a defensive safeguard, tested separately with damaged queued data.

@xxo1shine xxo1shine Nov 10, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I understand that an unmatched top-level END_GROUP is malformed and that adding ingress validation has a sound basis. However, this stall occurs because ingress accepts the data while deferred parsing rejects the same bytes. Reusing BlockMessage for deferred parsing would eliminate this difference while preserving the existing acceptance behavior, filtering, and contract-format checks.

I still recommend using the same parsing path in this PR to fix the regression with a minimal change.

If we instead reject the payload at ingress, it will never reach deferred parsing. In that case, adding the ingress check is sufficient to fix this reproduction, and I suggest removing the newly added state cleanup and BAD_BLOCK disconnect from the deferred-parsing catch block for now.

@317787106 317787106 Nov 10, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Updated in 7b1150b0b5.

Deferred decoding now uses new BlockMessage(unparsedBlock.getData()).getBlockCapsule(), matching the ingress path. The ingress end-tag check remains, so unmatched top-level END_GROUP payloads are still rejected under both filter settings.

Removed the unconditional BAD_BLOCK disconnect from the deferred catch. The processing-marker and request-reservation cleanup remains because deferred validation can fail after successful ingress: ALLOW_PROTO_FILTER_NUM may change while a block is queued, or waiting for contract validation may be interrupted. Cleanup allows retry without attributing these failures to the provider; a short code comment explains this.

The regression now exercises a filter change after enqueueing: it verifies that deferred rejection releases sync state without disconnecting the provider, then a canonical response from another peer is processed successfully.

Validation: 49 focused tests, clean build -x test, and the CI Checkstyle and Math checks passed.

@bladehan1 bladehan1 Nov 10, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[NIT] Add one line above this catch noting it is a defensive safeguard, now that ingress rejects trailing END_GROUP

With BlockMessage now rejecting unmatched top-level END_GROUP tags at ingress (f12c112), this deferred-parse failure path is no longer reachable via the wire for that payload: every queued UnparsedBlock carries the original bytes, and the ingress and deferred parses now accept the same language. That makes this cleanup-plus-disconnect look redundant to a reader (and it is the point xxo1shine is pressing on in this thread). It is still worth keeping as defense in depth — it guards against future queue-source changes or a parse-entry divergence silently re-opening the #6717 60-minute stall — but nothing in the code says so, so each future reviewer will re-derive the same debate.

Not blocking; a single line here would settle both the reachability question and the "is this dead code" concern.

Suggestion: Add one comment line above the catch, e.g.: "Ingress validation rejects malformed payloads; this cleanup is a defensive safeguard for queued data damaged after receipt or for future divergence between the two parse entries — do not remove without re-checking the #6717 stall recurrence."

@317787106 317787106 Nov 10, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added in 7b1150b0b5. The comment now explains that deferred validation may fail after successful ingress—for example, if ALLOW_PROTO_FILTER_NUM changes or waiting for contract validation is interrupted—and that releasing sync state allows retry.

Deferred decoding now also reuses BlockMessage. The unconditional BAD_BLOCK disconnect has been removed; the regression covers a filter change after enqueueing and verifies cleanup and retry without disconnecting the provider.

invalid(blockId, peerConnection);
return;
}
processSyncBlock(block, peerConnection);
Expand Down
55 changes: 55 additions & 0 deletions framework/src/test/java/org/tron/core/net/message/adv/BlockMessageTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,55 @@
package org.tron.core.net.message.adv;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThrows;
import static org.mockito.Mockito.CALLS_REAL_METHODS;
import static org.mockito.Mockito.mockStatic;

import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import java.util.Arrays;
import org.junit.Test;
import org.mockito.MockedStatic;
import org.tron.common.overlay.message.Message;
import org.tron.protos.Protocol.Block;
import org.tron.protos.Protocol.BlockHeader;

public class BlockMessageTest {

private static Block sampleBlock() {
return Block.newBuilder().setBlockHeader(BlockHeader.newBuilder()
.setRawData(BlockHeader.raw.newBuilder().setNumber(100).setTimestamp(123456789L))
.setWitnessSignature(ByteString.copyFrom(new byte[] {0x0c})))
.build();
}

@Test
public void trailingEndGroupIsRejectedWithEitherFilterSetting() {
byte[] canonical = sampleBlock().toByteArray();
for (boolean filter : new boolean[] {false, true}) {
try (MockedStatic<Message> messages = mockStatic(Message.class, CALLS_REAL_METHODS)) {
messages.when(Message::isFilter).thenReturn(filter);
// Also reject an END_GROUP followed by more bytes, not just a final 0x0c byte.
for (byte[] suffix : new byte[][] {{0x0c}, {0x0c, 0x08, 0x01}}) {
byte[] malformed = Arrays.copyOf(canonical, canonical.length + suffix.length);
System.arraycopy(suffix, 0, malformed, canonical.length, suffix.length);
assertThrows(InvalidProtocolBufferException.class, () -> new BlockMessage(malformed));
}
}
}
}

@Test
public void endGroupByteInsideABytesFieldIsAccepted() throws Exception {
Block block = sampleBlock();
byte[] data = block.toByteArray();
assertEquals(0x0c, data[data.length - 1]);
for (boolean filter : new boolean[] {false, true}) {
try (MockedStatic<Message> messages = mockStatic(Message.class, CALLS_REAL_METHODS)) {
messages.when(Message::isFilter).thenReturn(filter);
assertEquals(block, new BlockMessage(data).getBlockCapsule().getInstance());
}
}
}

}
6 changes: 5 additions & 1 deletion framework/src/test/java/org/tron/core/net/services/BlockFetchFailureTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.CALLS_REAL_METHODS;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
Expand Down Expand Up @@ -30,6 +31,7 @@
import org.mockito.MockedStatic;
import org.tron.common.TestConstants;
import org.tron.common.crypto.ECKey;
import org.tron.common.overlay.message.Message;
import org.tron.common.utils.ReflectUtils;
import org.tron.common.utils.Sha256Hash;
import org.tron.common.utils.Time;
Expand Down Expand Up @@ -447,7 +449,9 @@ private void assertSyncFailure(Exception failure, ReasonCode reason, boolean pre
Cache<BlockId, PeerConnection> requests =
(Cache<BlockId, PeerConnection>) ReflectUtils.getFieldObject(sync, "requestBlockIds");
requests.put(id, sender);
try {
try (MockedStatic<Message> messages = mockStatic(Message.class, CALLS_REAL_METHODS)) {
// These failure-policy tests do not initialize the global dynamic properties store.
messages.when(Message::isFilter).thenReturn(false);
sync.processBlock(sender, new BlockMessage(block));
invoke(sync, "handleSyncBlock");
verify(delegate).validBlock(any(BlockCapsule.class));
Expand Down
178 changes: 178 additions & 0 deletions framework/src/test/java/org/tron/core/net/services/SyncServiceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,10 @@
import static org.mockito.Mockito.mock;

import com.google.common.cache.Cache;
import com.google.protobuf.ByteString;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.math.BigInteger;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
Expand All @@ -13,21 +15,26 @@
import java.util.Map;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.springframework.context.ApplicationContext;
import org.tron.common.BaseMethodTest;
import org.tron.common.crypto.ECKey;
import org.tron.common.overlay.message.Message;
import org.tron.common.utils.ReflectUtils;
import org.tron.common.utils.Sha256Hash;
import org.tron.core.capsule.BlockCapsule;
import org.tron.core.net.P2pEventHandlerImpl;
import org.tron.core.net.TronNetDelegate;
import org.tron.core.net.message.MessageTypes;
import org.tron.core.net.message.adv.BlockMessage;
import org.tron.core.net.message.adv.FetchInvDataMessage;
import org.tron.core.net.messagehandler.BlockMsgHandler;
import org.tron.core.net.messagehandler.PbftDataSyncHandler;
import org.tron.core.net.peer.PeerConnection;
import org.tron.core.net.peer.PeerManager;
import org.tron.core.net.peer.TronState;
import org.tron.core.net.service.adv.AdvService;
import org.tron.core.net.service.sync.SyncService;
import org.tron.core.net.service.sync.UnparsedBlock;
import org.tron.p2p.connection.Channel;
Expand Down Expand Up @@ -270,6 +277,177 @@ public void testProcessingBlockResumesThrottledFetch() throws Exception {
sync.close();
}

@Test
public void testStoredGenesisBlockRoundTripsThroughNetworkDecoder() throws Exception {
BlockCapsule stored = chainBaseManager.getBlockById(chainBaseManager.getGenesisBlockId());
for (int filter : new int[] {0, 1}) {
chainBaseManager.getDynamicPropertiesStore().saveAllowProtoFilterNum(filter);
BlockMessage message = new BlockMessage(new BlockMessage(stored).getData());
Assert.assertEquals(stored.getInstance(), message.getBlockCapsule().getInstance());
Assert.assertEquals(stored.getBlockId(), message.getBlockId());
Assert.assertArrayEquals(stored.getData(), message.getData());
}
}

@Test
public void testIngressParseFailureAllowsRetryAfterDisconnect() throws Exception {
chainBaseManager.getDynamicPropertiesStore().saveAllowProtoFilterNum(0);
SyncService sync = new SyncService();
try {
TronNetDelegate delegate = mock(TronNetDelegate.class);
ReflectUtils.setFieldValue(sync, "tronNetDelegate", delegate);
PeerConnection badPeer = Mockito.spy(context.getBean(PeerConnection.class));
PeerConnection goodPeer = Mockito.spy(context.getBean(PeerConnection.class));
Channel channel = mock(Channel.class);
Channel goodChannel = mock(Channel.class);
Mockito.when(channel.getInetSocketAddress()).thenReturn(inetSocketAddress);
Mockito.when(channel.getInetAddress()).thenReturn(inetSocketAddress.getAddress());
Mockito.when(goodChannel.getInetSocketAddress()).thenReturn(inetSocketAddress);
Mockito.when(goodChannel.getInetAddress()).thenReturn(inetSocketAddress.getAddress());
badPeer.setChannel(channel);
goodPeer.setChannel(goodChannel);
badPeer.setFetchAble(true);
goodPeer.setFetchAble(true);
ReflectUtils.setFieldValue(badPeer, "syncService", sync);
ReflectUtils.setFieldValue(badPeer, "advService", mock(AdvService.class));
Mockito.when(delegate.getActivePeer()).thenReturn(Arrays.asList(badPeer, goodPeer));

BlockCapsule block = new BlockCapsule(1, Sha256Hash.ZERO_HASH,
System.currentTimeMillis(),
ByteString.copyFrom(ECKey.fromPrivate(BigInteger.ONE).getAddress()));
BlockCapsule.BlockId blockId = block.getBlockId();
badPeer.getSyncBlockToFetch().add(blockId);
goodPeer.getSyncBlockToFetch().add(blockId);
Method fetch = SyncService.class.getDeclaredMethod("startFetchSyncBlock");
fetch.setAccessible(true);
fetch.invoke(sync);
Cache<BlockCapsule.BlockId, PeerConnection> requests =
(Cache<BlockCapsule.BlockId, PeerConnection>)
ReflectUtils.getFieldObject(sync, "requestBlockIds");
Assert.assertSame(badPeer, requests.getIfPresent(blockId));

byte[] data = block.getData();
byte[] wire = new byte[data.length + 2];
wire[0] = MessageTypes.BLOCK.asByte();
System.arraycopy(data, 0, wire, 1, data.length);
wire[wire.length - 1] = 0x0c;
P2pEventHandlerImpl handler = new P2pEventHandlerImpl();
BlockMsgHandler blockHandler = mock(BlockMsgHandler.class);
ReflectUtils.setFieldValue(handler, "blockMsgHandler", blockHandler);
Method process = P2pEventHandlerImpl.class
.getDeclaredMethod("processMessage", PeerConnection.class, byte[].class);
process.setAccessible(true);
process.invoke(handler, badPeer, wire);

Mockito.verifyNoInteractions(blockHandler);
Mockito.verify(badPeer).disconnect(Mockito.any(Protocol.ReasonCode.class));
Assert.assertTrue(badPeer.getSyncBlockRequested().containsKey(blockId));
Assert.assertTrue(badPeer.getSyncBlockInProcess().isEmpty());
badPeer.onDisconnect();
Assert.assertNull(requests.getIfPresent(blockId));
Assert.assertTrue(badPeer.getSyncBlockRequested().isEmpty());
Mockito.when(delegate.getActivePeer()).thenReturn(Collections.singletonList(goodPeer));
fetch.invoke(sync);
Assert.assertSame(goodPeer, requests.getIfPresent(blockId));
Assert.assertTrue(goodPeer.getSyncBlockRequested().containsKey(blockId));
Mockito.verify(goodPeer).sendMessage(Mockito.any(FetchInvDataMessage.class));
} finally {
sync.close();
}
}

@Test
public void testDeserializeFailureAllowsRetryFromAnotherPeer() throws Exception {

@bladehan1 bladehan1 Nov 6, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[NIT] Declare the non-singleton isolation premise for these tests (class-level javadoc)

This test (and the other new SyncService()-based tests in this class) constructs the system under test as a plain local instance and mutates it only via reflection on instance fields; PeerConnection is a prototype-scoped bean, so no Spring singleton or static mutable state is touched. That makes the isolation premise sound, but per the project's test-singleton hygiene rule (H7) the premise must be stated explicitly in the test class javadoc — otherwise every future reviewer has to re-derive it from scratch.

Suggestion: add one line to the class-level javadoc of SyncServiceTest stating that tests constructing fresh SyncService instances share no Spring singleton state and therefore need no @After reset.

@317787106 317787106 Nov 9, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

SyncServiceTest extends BaseMethodTest, which creates a fresh Spring context for each test method and provides an inherited @After that closes the context and clears Args. The locally constructed SyncService instances are also closed in finally blocks.

That lifecycle is already documented in BaseMethodTest. Some tests here use Spring beans and modify dynamic properties, so a class-level statement that no singleton state is touched would be inaccurate. No additional @After is needed for these additions.

chainBaseManager.getDynamicPropertiesStore().saveAllowProtoFilterNum(0);
SyncService sync = new SyncService();
try {
TronNetDelegate delegate = mock(TronNetDelegate.class);
ReflectUtils.setFieldValue(sync, "tronNetDelegate", delegate);
ReflectUtils.setFieldValue(sync, "pbftDataSyncHandler", mock(PbftDataSyncHandler.class));
Mockito.when(delegate.getSolidBlockId())
.thenReturn(new BlockCapsule.BlockId(Sha256Hash.ZERO_HASH, 0));
Mockito.when(delegate.getBlockLock()).thenReturn(new Object());

PeerConnection badPeer = Mockito.spy(context.getBean(PeerConnection.class));
PeerConnection goodPeer = Mockito.spy(context.getBean(PeerConnection.class));
Channel badChannel = mock(Channel.class);
Channel goodChannel = mock(Channel.class);
Mockito.when(badChannel.getInetSocketAddress()).thenReturn(inetSocketAddress);
Mockito.when(badChannel.getInetAddress()).thenReturn(inetSocketAddress.getAddress());
Mockito.when(goodChannel.getInetSocketAddress()).thenReturn(inetSocketAddress);
Mockito.when(goodChannel.getInetAddress()).thenReturn(inetSocketAddress.getAddress());
badPeer.setChannel(badChannel);
goodPeer.setChannel(goodChannel);
badPeer.setFetchAble(true);
goodPeer.setFetchAble(true);
Mockito.when(delegate.getActivePeer()).thenReturn(Arrays.asList(badPeer, goodPeer));

BlockCapsule block = new BlockCapsule(1, Sha256Hash.ZERO_HASH,
System.currentTimeMillis(),
ByteString.copyFrom(ECKey.fromPrivate(BigInteger.ONE).getAddress()));
BlockCapsule.BlockId blockId = block.getBlockId();
byte[] data = block.getData();
// Repeating the singular header field preserves the block but changes its wire length.
byte[] nonCanonical = Arrays.copyOf(data, data.length * 2);
System.arraycopy(data, 0, nonCanonical, data.length, data.length);
BlockMessage message = new BlockMessage(nonCanonical);
Assert.assertEquals(block.getInstance(), message.getBlockCapsule().getInstance());
badPeer.getSyncBlockToFetch().add(blockId);
goodPeer.getSyncBlockToFetch().add(blockId);

Method fetch = SyncService.class.getDeclaredMethod("startFetchSyncBlock");
fetch.setAccessible(true);
fetch.invoke(sync);
Assert.assertTrue(badPeer.getSyncBlockRequested().containsKey(blockId));
Assert.assertTrue(goodPeer.getSyncBlockRequested().isEmpty());
Cache<BlockCapsule.BlockId, PeerConnection> requests =
(Cache<BlockCapsule.BlockId, PeerConnection>)
ReflectUtils.getFieldObject(sync, "requestBlockIds");
Assert.assertSame(badPeer, requests.getIfPresent(blockId));

BlockMsgHandler handler = new BlockMsgHandler();
ReflectUtils.setFieldValue(handler, "syncService", sync);
handler.processMessage(badPeer, message);
Assert.assertTrue(badPeer.getSyncBlockRequested().isEmpty());
Assert.assertTrue(badPeer.getSyncBlockInProcess().contains(blockId));
// A filter change while the block is queued can make deferred validation fail.
chainBaseManager.getDynamicPropertiesStore().saveAllowProtoFilterNum(1);
Method handle = SyncService.class.getDeclaredMethod("handleSyncBlock");
handle.setAccessible(true);
ReflectUtils.setFieldValue(sync, "fetchFlag", false);
handle.invoke(sync);

Assert.assertNull("A failed parse must release the global request",
requests.getIfPresent(blockId));
Assert.assertFalse(badPeer.getSyncBlockInProcess().contains(blockId));
Mockito.verify(badPeer, Mockito.never()).disconnect(Mockito.any(Protocol.ReasonCode.class));
Mockito.verify(delegate, Mockito.never())
.processBlock(Mockito.any(BlockCapsule.class), Mockito.anyBoolean());
Assert.assertTrue((boolean) ReflectUtils.getFieldObject(sync, "fetchFlag"));

Mockito.when(delegate.getActivePeer()).thenReturn(Collections.singletonList(goodPeer));
fetch.invoke(sync);
ArgumentCaptor<FetchInvDataMessage> request =
ArgumentCaptor.forClass(FetchInvDataMessage.class);
Mockito.verify(goodPeer).sendMessage(request.capture());
Assert.assertEquals(Collections.singletonList(blockId), request.getValue().getHashList());
Assert.assertSame(goodPeer, requests.getIfPresent(blockId));

// Leave a next block queued so successful processing does not request a chain summary.
BlockCapsule.BlockId next = new BlockCapsule.BlockId(Sha256Hash.ZERO_HASH, 2);
goodPeer.getSyncBlockToFetch().add(next);
handler.processMessage(goodPeer, new BlockMessage(block));
handle.invoke(sync);
Mockito.verify(delegate).processBlock(Mockito.any(BlockCapsule.class), Mockito.eq(true));
Assert.assertEquals(blockId, goodPeer.getBlockBothHave());
Assert.assertEquals(next, goodPeer.getSyncBlockToFetch().peek());
Assert.assertFalse(goodPeer.getSyncBlockInProcess().contains(blockId));
Mockito.verify(goodPeer, Mockito.never()).disconnect(Mockito.any(Protocol.ReasonCode.class));
} finally {
sync.close();
}
}

@Test
public void sameAddressPeersKeepSyncRequestsAndResponsesIndependent() throws Exception {
InetSocketAddress address = new InetSocketAddress("127.0.0.2", 10001);
Expand Down
Loading