feat: end the stream with TOO_FAR_BEHIND if the block is missing from the buffer (#21912)

Signed-off-by: Neeharika-Sompalli <[email protected]>
Signed-off-by: Petar Tonev <[email protected]>
Signed-off-by: Zhivko Kelchev <[email protected]>
Signed-off-by: mxtartaglia <[email protected]>
Signed-off-by: Matt Hess <[email protected]>
Signed-off-by: Anthony Petrov <[email protected]>
Signed-off-by: Jendrik Johannes <[email protected]>
Signed-off-by: Derek Riley <[email protected]>
Signed-off-by: Tim Farber-Newman <[email protected]>
Signed-off-by: Michael Heinrichs <[email protected]>
Signed-off-by: Michael Tinker <[email protected]>
Signed-off-by: Artur Biesiadowski <[email protected]>
Signed-off-by: Josh Marinacci <[email protected]>
Signed-off-by: Roger Barker <[email protected]>
Signed-off-by: ibankov <[email protected]>
Signed-off-by: Kelly Greco <[email protected]>
Signed-off-by: Eva <[email protected]>
Co-authored-by: Neeharika Sompalli <[email protected]>
Co-authored-by: Zhivko Kelchev <[email protected]>
Co-authored-by: Maxi Tartaglia <[email protected]>
Co-authored-by: Matt Hess <[email protected]>
Co-authored-by: anthony-swirldslabs <[email protected]>
Co-authored-by: Jendrik Johannes <[email protected]>
Co-authored-by: Derek Riley <[email protected]>
Co-authored-by: Tim Farber-Newman <[email protected]>
Co-authored-by: Michael Heinrichs <[email protected]>
Co-authored-by: Michael Tinker <[email protected]>
Co-authored-by: Artur Biesiadowski <[email protected]>
Co-authored-by: Josh Marinacci <[email protected]>
Co-authored-by: Roger Barker <[email protected]>
Co-authored-by: Ivan Bankov <[email protected]>
Co-authored-by: Kelly Greco <[email protected]>
Co-authored-by: Evdokia-Georgieva <[email protected]>

Author: 17 people

hedera-node/hedera-app/src/main/java/com/hedera/node/app/blocks/impl/streaming/BlockBufferService.java

@@ -334,7 +334,7 @@ public void openBlock(final long blockNumber) {
         // Create a new block state
         final BlockState blockState = new BlockState(blockNumber);
         blockBuffer.put(blockNumber, blockState);
-        // update the earliest block number if this is first block or lower than current earliest
+        // update the earliest block number if this is the first block or lower than current earliest
         earliestBlockNumber.updateAndGet(
                 current -> current == Long.MIN_VALUE ? blockNumber : Math.min(current, blockNumber));
         lastProducedBlockNumber.updateAndGet(old -> Math.max(old, blockNumber));
@@ -592,7 +592,7 @@ block period (e.g. 2 seconds). This gives us an ideal number of blocks in the bu
                     if (block.blockNumber() > highestBlockAcked) {
                         ++numPendingAck;
                     }
-                    // Keep track of earliest remaining block
+                    // Keep track of the earliest and the latest remaining blocks
                     newEarliestBlock = Math.min(newEarliestBlock, blockNum);
                     newLatestBlock = Math.max(newLatestBlock, blockNum);
                 }
@@ -602,13 +602,13 @@ block period (e.g. 2 seconds). This gives us an ideal number of blocks in the bu
                     it.remove();
                     ++numPruned;
                 } else {
-                    // keep track of earliest remaining block
+                    // Keep track of the earliest and the latest remaining blocks
                     newEarliestBlock = Math.min(newEarliestBlock, blockNum);
                     newLatestBlock = Math.max(newLatestBlock, blockNum);
                 }
             } else {
                 ++numPendingAck;
-                // keep track of earliest remaining block
+                // Keep track of the earliest and the latest remaining blocks
                 newEarliestBlock = Math.min(newEarliestBlock, blockNum);
                 newLatestBlock = Math.max(newLatestBlock, blockNum);
             }

hedera-node/hedera-app/src/main/java/com/hedera/node/app/blocks/impl/streaming/BlockNodeConnection.java

@@ -2,6 +2,7 @@
 package com.hedera.node.app.blocks.impl.streaming;
 
 import static java.util.Objects.requireNonNull;
+import static org.hiero.block.api.PublishStreamRequest.EndStream.Code.ERROR;
 import static org.hiero.block.api.PublishStreamRequest.EndStream.Code.RESET;
 import static org.hiero.block.api.PublishStreamRequest.EndStream.Code.TIMEOUT;
 import static org.hiero.block.api.PublishStreamRequest.EndStream.Code.TOO_FAR_BEHIND;
@@ -612,7 +613,13 @@ private void handleResendBlock(@NonNull final ResendBlock resendBlock) {
                             + "on this consensus node. Closing connection and will retry later.",
                     this,
                     resendBlockNumber);
-            closeAndReschedule(THIRTY_SECONDS, true);
+
+            if (resendBlockNumber < blockBufferService.getEarliestAvailableBlockNumber()) {
+                // Indicate that the block node should catch up from another trustworthy block node
+                endStreamAndReschedule(TOO_FAR_BEHIND);
+            } else if (resendBlockNumber > blockBufferService.getLastBlockNumberProduced()) {
+                endStreamAndReschedule(ERROR);
+            }
         }
     }
 

hedera-node/hedera-app/src/test/java/com/hedera/node/app/blocks/impl/streaming/BlockNodeCommunicationTestBase.java

@@ -81,6 +81,15 @@ protected static PublishStreamRequest createRequest(final EndStream.Code endCode
         return PublishStreamRequest.newBuilder().endStream(endStream).build();
     }
 
+    @NonNull
+    protected static PublishStreamRequest createRequest(final EndStream.Code endCode, final long earliestBlockNumber) {
+        final EndStream endStream = EndStream.newBuilder()
+                .endCode(endCode)
+                .earliestBlockNumber(earliestBlockNumber)
+                .build();
+        return PublishStreamRequest.newBuilder().endStream(endStream).build();
+    }
+
     @NonNull
     protected static PublishStreamRequest createRequest(final long blockNumber) {
         final BlockEnd endOfBlock =

hedera-node/hedera-app/src/test/java/com/hedera/node/app/blocks/impl/streaming/BlockNodeConnectionTest.java

@@ -592,13 +592,14 @@ void testOnNext_resendBlock_blockExists() {
     }
 
     @Test
-    void testOnNext_resendBlock_blockDoesNotExist() {
+    void testOnNext_resendBlock_blockDoesNotExist_TooFarBehind() {
         openConnectionAndResetMocks();
 
         final AtomicLong streamingBlockNumber = streamingBlockNumber();
         streamingBlockNumber.set(11); // pretend we are currently streaming block 11
         final PublishStreamResponse response = createResendBlock(10L);
         when(bufferService.getBlockState(10L)).thenReturn(null);
+        when(bufferService.getEarliestAvailableBlockNumber()).thenReturn(11L);
         connection.updateConnectionState(ConnectionState.ACTIVE);
 
         connection.onNext(response);
@@ -607,9 +608,46 @@ void testOnNext_resendBlock_blockDoesNotExist() {
         verify(metrics).recordResponseReceived(ResponseOneOfType.RESEND_BLOCK);
         verify(metrics).recordConnectionClosed();
         verify(metrics).recordActiveConnectionIp(-1L);
+        verify(metrics).recordRequestLatency(anyLong());
+        verify(metrics).recordRequestEndStreamSent(EndStream.Code.TOO_FAR_BEHIND);
+        verify(requestPipeline).onNext(createRequest(EndStream.Code.TOO_FAR_BEHIND, 11L));
         verify(requestPipeline).onComplete();
         verify(connectionManager).rescheduleConnection(connection, Duration.ofSeconds(30), null, true);
+        verify(connectionManager).notifyConnectionClosed(connection);
         verify(bufferService).getBlockState(10L);
+        verify(bufferService, times(2)).getEarliestAvailableBlockNumber();
+        verify(bufferService).getHighestAckedBlockNumber();
+        verifyNoMoreInteractions(metrics);
+        verifyNoMoreInteractions(requestPipeline);
+        verifyNoMoreInteractions(connectionManager);
+        verifyNoMoreInteractions(bufferService);
+    }
+
+    @Test
+    void testOnNext_resendBlock_blockDoesNotExist_Error() {
+        openConnectionAndResetMocks();
+
+        final AtomicLong streamingBlockNumber = streamingBlockNumber();
+        streamingBlockNumber.set(11); // pretend we are currently streaming block 11
+        final PublishStreamResponse response = createResendBlock(10L);
+        when(bufferService.getBlockState(10L)).thenReturn(null);
+        connection.updateConnectionState(ConnectionState.ACTIVE);
+
+        connection.onNext(response);
+
+        verify(metrics).recordLatestBlockResendBlock(10L);
+        verify(metrics).recordResponseReceived(ResponseOneOfType.RESEND_BLOCK);
+        verify(metrics).recordConnectionClosed();
+        verify(metrics).recordActiveConnectionIp(-1L);
+        verify(metrics).recordRequestLatency(anyLong());
+        verify(metrics).recordRequestEndStreamSent(EndStream.Code.ERROR);
+        verify(requestPipeline).onNext(createRequest(EndStream.Code.ERROR));
+        verify(requestPipeline).onComplete();
+        verify(connectionManager).rescheduleConnection(connection, Duration.ofSeconds(30), null, true);
+        verify(bufferService).getBlockState(10L);
+        verify(bufferService, times(2)).getEarliestAvailableBlockNumber();
+        verify(bufferService).getLastBlockNumberProduced();
+        verify(bufferService).getHighestAckedBlockNumber();
         verifyNoMoreInteractions(metrics);
         verifyNoMoreInteractions(requestPipeline);
         verifyNoMoreInteractions(bufferService);

hedera-node/test-clients/src/main/java/com/hedera/services/bdd/suites/blocknode/BlockNodeSuite.java

@@ -919,8 +919,7 @@ final Stream<DynamicTest> testCNReactionToPublishStreamResponses() {
                                 portNumbers.getFirst()),
                         String.format(
                                 "/localhost:%s/ACTIVE] Block node requested a ResendBlock for block 9223372036854775807 but that block does not exist on this consensus node. Closing connection and will retry later",
-                                portNumbers.getFirst()))),
-                assertBlockNodeCommsLogDoesNotContain(byNodeId(0), "ERROR", Duration.ofSeconds(5)));
+                                portNumbers.getFirst()))));
     }
 
     @NotNull