fix(tools): wait for all ACKs before closing send side

Signed-off-by: Alfredo Gutierrez <alfredo@hashgraph.com>

Author: AlfredoG87

tools-and-tests/tools/src/main/java/org/hiero/block/tools/capacity/NetworkCapacityClient.java

@@ -100,7 +100,8 @@ private BlockStreamPublishClient createPublishClient() {
 
     private void streamBlocksFromFolder(BlockStreamPublishClient publishClient) throws IOException {
         List<Path> blockFiles = Files.list(recordingFolder)
-                .filter(path -> path.toString().endsWith(".blk.gz") || path.toString().endsWith(".blk"))
+                .filter(path ->
+                        path.toString().endsWith(".blk.gz") || path.toString().endsWith(".blk"))
                 .sorted()
                 .collect(Collectors.toList());
 
@@ -152,6 +153,12 @@ private void streamBlocksFromFolder(BlockStreamPublishClient publishClient) thro
             metrics.reportPeriodic();
         }
 
+        // Wait for all ACKs to arrive before half-closing the send side.
+        // The last ACK is in-flight from the server while we finish the send loop.
+        long ackDeadline = System.currentTimeMillis() + 2_000;
+        while (metrics.getTotalBlocksAcked() < metrics.getTotalBlocks() && System.currentTimeMillis() < ackDeadline) {
+            LockSupport.parkNanos(10_000_000L);
+        }
         requestPipeline.onComplete();
 
         try {

tools-and-tests/tools/src/main/java/org/hiero/block/tools/capacity/NetworkCapacityServer.java

@@ -139,8 +139,7 @@ public void onNext(final PublishStreamRequestUnparsed req) {
                 if (items.getLast().hasBlockProof() && hasReceivedBlockHeader) {
                     metrics.incrementBlocks();
                     System.out.printf(
-                            "[SERVER] Completed block %d (Total %d).%n",
-                            currentBlockNumber, metrics.getTotalBlocks());
+                            "[SERVER] Completed block %d (Total %d).%n", currentBlockNumber, metrics.getTotalBlocks());
 
                     PublishStreamResponse response = PublishStreamResponse.newBuilder()
                             .acknowledgement(PublishStreamResponse.BlockAcknowledgement.newBuilder()