initial refactor of batching logic

Signed-off-by: Atanas Atanasov <a.v.atanasov98@gmail.com>

Author: ata-nas

block-node/app/src/testFixtures/java/org/hiero/block/node/app/fixtures/blocks/InMemoryBlockAccessor.java

@@ -0,0 +1,61 @@
+// SPDX-License-Identifier: Apache-2.0
+package org.hiero.block.node.app.fixtures.blocks;
+
+import com.hedera.hapi.block.stream.Block;
+import com.hedera.hapi.block.stream.BlockItem;
+import java.util.List;
+import java.util.Objects;
+import org.hiero.block.node.spi.historicalblocks.BlockAccessor;
+
+/**
+ * A simple in-memory block accessor class. This class is intended to be used
+ * for testing purposes only!
+ */
+public final class InMemoryBlockAccessor implements BlockAccessor {
+    /** Internal in-memory reference of the accessible block */
+    private final Block block;
+    /** The block number of the accessible block */
+    private final long blockNumber;
+
+    /**
+     * Constructor.
+     *
+     * @param blockItems of the block, must not be empty, first item must be a
+     * {@link com.hedera.hapi.block.stream.output.BlockHeader} and last item
+     * must be a {@link com.hedera.hapi.block.stream.BlockProof}.
+     * @throws IllegalArgumentException if preconditions are not met
+     */
+    public InMemoryBlockAccessor(final List<BlockItem> blockItems) {
+        if (blockItems.isEmpty()) {
+            throw new IllegalArgumentException("Block items list cannot be empty");
+        }
+        // Ensure the first item is a BlockHeader
+        if (!blockItems.getFirst().hasBlockHeader()) {
+            throw new IllegalArgumentException("First block item must be a BlockHeader");
+        }
+        // Ensure the last item is a BlockProof
+        if (!blockItems.getLast().hasBlockProof()) {
+            throw new IllegalArgumentException("Last block item must be a BlockProof");
+        }
+        // Create a Block from the provided block items
+        block = Block.newBuilder().items(blockItems).build();
+        blockNumber =
+                Objects.requireNonNull(block.items().getFirst().blockHeader()).number();
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public long blockNumber() {
+        return blockNumber;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public Block block() {
+        return block;
+    }
+}

block-node/block-providers/files.historic/src/main/java/org/hiero/block/node/blocks/files/historic/BlocksFilesHistoricPlugin.java

@@ -8,6 +8,7 @@
 import java.io.IOException;
 import java.lang.System.Logger.Level;
 import java.nio.file.Files;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -21,6 +22,7 @@
 import org.hiero.block.node.spi.historicalblocks.BlockAccessor;
 import org.hiero.block.node.spi.historicalblocks.BlockProviderPlugin;
 import org.hiero.block.node.spi.historicalblocks.BlockRangeSet;
+import org.hiero.block.node.spi.historicalblocks.HistoricalBlockFacility;
 import org.hiero.block.node.spi.historicalblocks.LongRange;
 
 /**
@@ -195,9 +197,13 @@ public BlockRangeSet availableBlocks() {
      */
     @Override
     public void handlePersisted(PersistedNotification notification) {
+        // all operations here until we release the caller thread are queued up
+        // by the block messaging facility, so we can safely perform
+        // calculations and decide if we will submit or not a task
         if (notification.blockProviderPriority() > defaultPriority()) {
             attemptZipping();
-        } // todo this is not enough of an assertion that the blocks will be coming from the right place
+        }
+        // @todo(1069) this is not enough of an assertion that the blocks will be coming from the right place
         //     as notifications are async and things can happen, when we get the accessors later, we should
         //     be able to get accessors only from places that have higher priority than us. We should probably
         //     have that as a feature in the block accessor api. (meaning we should be able to query the
@@ -207,12 +213,20 @@ public void handlePersisted(PersistedNotification notification) {
     // ==== Private Methods ============================================================================================
     private void attemptZipping() {
         // compute the min and max block in next batch to zip
-        long minBlockNumber = availableBlocks().max() + 1;
+        // since we ensure no gaps in the zip file are possible, and also we
+        // have a power of 10 number of blocks per zip file, it is safe to
+        // simply add +1 to the latest available block number and have that as
+        // the start of the next batch of blocks to zip
+        long minBlockNumber = availableBlocks.max() + 1;
         long maxBlockNumber = minBlockNumber + numberOfBlocksPerZipFile - 1;
-        // check all those blocks are available
+        // make sure the historical block facility has a higher max than our
+        // desired batch max
         final BlockRangeSet historicalAvailable =
                 context.historicalBlockProvider().availableBlocks();
         // while we can zip blocks, we must keep zipping
+        // we loop here because the historical block facility can have
+        // multiple batches of blocks available for zipping potentially, so we
+        // need to queue them all up
         while (historicalAvailable.max() >= maxBlockNumber) {
             if (historicalAvailable.contains(minBlockNumber, maxBlockNumber)) {
                 final LongRange batchRange = new LongRange(minBlockNumber, maxBlockNumber);
@@ -234,17 +248,19 @@ private void attemptZipping() {
     private void startMovingBatchOfBlocksToZipFile(final LongRange batchRange) {
         // check if the batch of blocks is already in progress
         if (inProgressZipRanges.contains(batchRange)) {
+            // if the batch is in progress, we must not submit a task
             LOGGER.log(
                     System.Logger.Level.DEBUG,
                     "Batch of blocks[{%1} -> {%2}] is already in progress",
                     batchRange.start(),
                     batchRange.end());
-            return;
+        } else {
+            // if the batch is not in progress, we must submit a task
+            // add the batch of blocks to the in progress ranges
+            inProgressZipRanges.add(batchRange);
+            // move the batch of blocks to a zip file (submit a task)
+            zipMoveExecutorService.submit(() -> moveBatchOfBlocksToZipFile(batchRange));
         }
-        // add the batch of blocks to the in progress ranges
-        inProgressZipRanges.add(batchRange);
-        // move the batch of blocks to a zip file
-        zipMoveExecutorService.submit(() -> moveBatchOfBlocksToZipFile(batchRange));
     }
 
     /**
@@ -253,47 +269,70 @@ private void startMovingBatchOfBlocksToZipFile(final LongRange batchRange) {
      * @param batchRange The range of blocks to move to zip file.
      */
     private void moveBatchOfBlocksToZipFile(final LongRange batchRange) {
-        final long batchFirstBlockNumber = batchRange.start();
-        final long batchLastBlockNumber = batchRange.end();
-        // move the batch of blocks to a zip file
         try {
-            LOGGER.log(
-                    System.Logger.Level.DEBUG,
-                    "Moving batch of blocks[%d -> %d] to zip file",
-                    batchFirstBlockNumber,
-                    batchLastBlockNumber);
-
-            // Write the zip file and get result with file size
-            final long zipFileSize = zipBlockArchive.writeNewZipFile(batchFirstBlockNumber);
-            // Metrics updates
-            // Update total bytes stored with the new zip file size
-            totalBytesStored.addAndGet(zipFileSize);
-            // Increment the blocks written counter
-            long blockCount = batchLastBlockNumber - batchFirstBlockNumber + 1;
-            blocksWrittenCounter.add(blockCount);
-        } catch (final IOException e) {
-            LOGGER.log(
-                    System.Logger.Level.ERROR,
-                    "Failed to move batch of blocks[" + batchFirstBlockNumber + " -> " + batchLastBlockNumber
-                            + "] to zip file",
-                    e);
-            return;
+            // first off, let's create our batch of blocks
+            final long batchFirstBlockNumber = batchRange.start();
+            final long batchLastBlockNumber = batchRange.end();
+            final List<BlockAccessor> batch = new ArrayList<>(numberOfBlocksPerZipFile);
+            final HistoricalBlockFacility historicalBlockFacility = context.historicalBlockProvider();
+            // gather batch, if there are no gaps, then we can proceed with zipping
+            for (long blockNumber = batchFirstBlockNumber; blockNumber <= batchLastBlockNumber; blockNumber++) {
+                final BlockAccessor currentAccessor = historicalBlockFacility.block(blockNumber);
+                if (currentAccessor == null) {
+                    break;
+                } else {
+                    batch.add(currentAccessor);
+                }
+            }
+            // if there are any gaps, then we cannot zip the batch
+            if (batch.size() != numberOfBlocksPerZipFile) {
+                // we have a gap in the batch, so we cannot zip it
+                LOGGER.log(
+                        System.Logger.Level.DEBUG,
+                        "Batch of blocks[%d -> %d] has a gap, skipping zipping",
+                        batchFirstBlockNumber,
+                        batchLastBlockNumber);
+            } else {
+                // move the batch of blocks to a zip file
+                try {
+                    LOGGER.log(
+                            System.Logger.Level.DEBUG,
+                            "Moving batch of blocks[%d -> %d] to zip file",
+                            batchFirstBlockNumber,
+                            batchLastBlockNumber);
+                    // Write the zip file and get result with file size
+                    final long zipFileSize = zipBlockArchive.writeNewZipFile(batch);
+                    // Metrics updates
+                    // Update total bytes stored with the new zip file size
+                    totalBytesStored.addAndGet(zipFileSize);
+                    // Increment the blocks written counter
+                    blocksWrittenCounter.add(numberOfBlocksPerZipFile);
+                } catch (final IOException e) {
+                    LOGGER.log(
+                            System.Logger.Level.ERROR,
+                            "Failed to move batch of blocks[" + batchFirstBlockNumber + " -> " + batchLastBlockNumber
+                                    + "] to zip file",
+                            e);
+                    return;
+                }
+                // if we have reached here, then the batch of blocks has been zipped,
+                // now we need to make some updates
+                // update the first and last block numbers
+                availableBlocks.add(batchFirstBlockNumber, batchLastBlockNumber);
+                // log done
+                LOGGER.log(
+                        Level.INFO,
+                        "Moved batch of blocks[%d -> %d] to zip file",
+                        batchFirstBlockNumber,
+                        batchLastBlockNumber);
+                // now all the blocks are in the zip file and accessible, send notification
+                context.blockMessaging()
+                        .sendBlockPersisted(new PersistedNotification(
+                                batchFirstBlockNumber, batchLastBlockNumber, defaultPriority()));
+            }
+        } finally {
+            // always make sure to remove the batch of blocks from in progress ranges
+            inProgressZipRanges.remove(batchRange);
         }
-        // if we have reached here, then the batch of blocks has been zipped,
-        // now we need to make some updates
-        // update the first and last block numbers
-        availableBlocks.add(batchFirstBlockNumber, batchLastBlockNumber);
-        // log done
-        LOGGER.log(
-                System.Logger.Level.INFO,
-                "Moved batch of blocks[%d -> %d] to zip file",
-                batchFirstBlockNumber,
-                batchLastBlockNumber);
-        // now all the blocks are in the zip file and accessible, send notification
-        context.blockMessaging()
-                .sendBlockPersisted(
-                        new PersistedNotification(batchFirstBlockNumber, batchLastBlockNumber, defaultPriority()));
-        // remove the batch of blocks from in progress ranges
-        inProgressZipRanges.remove(batchRange);
     }
 }

block-node/block-providers/files.historic/src/main/java/org/hiero/block/node/blocks/files/historic/ZipBlockArchive.java

@@ -21,16 +21,15 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
-import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import java.util.zip.CRC32;
+import java.util.zip.Deflater;
 import java.util.zip.ZipEntry;
 import java.util.zip.ZipOutputStream;
 import org.hiero.block.node.base.BlockFile;
 import org.hiero.block.node.spi.BlockNodeContext;
 import org.hiero.block.node.spi.historicalblocks.BlockAccessor;
 import org.hiero.block.node.spi.historicalblocks.BlockAccessor.Format;
-import org.hiero.block.node.spi.historicalblocks.HistoricalBlockFacility;
 
 /**
  * The ZipBlockArchive class provides methods for creating and managing zip files containing blocks.
@@ -43,69 +42,50 @@ class ZipBlockArchive {
     private final BlockNodeContext context;
     /** The configuration for the historic files. */
     private final FilesHistoricConfig config;
-    /** The historical block facility. */
-    private final HistoricalBlockFacility historicalBlockFacility;
-    /** The number of blocks per zip file. */
-    private final int numberOfBlocksPerZipFile;
     /** The format for the blocks. */
     private final Format format;
 
     /**
      * Constructor for ZipBlockArchive.
      *
-     * @param context The block node context
      * @param filesHistoricConfig Configuration to be used internally
      */
     ZipBlockArchive(@NonNull final BlockNodeContext context, @NonNull final FilesHistoricConfig filesHistoricConfig) {
         this.context = Objects.requireNonNull(context);
         this.config = Objects.requireNonNull(filesHistoricConfig);
-        this.historicalBlockFacility = Objects.requireNonNull(context.historicalBlockProvider());
-        numberOfBlocksPerZipFile = (int) Math.pow(10, this.config.powersOfTenPerZipFileContents());
         format = switch (this.config.compression()) {
             case ZSTD -> Format.ZSTD_PROTOBUF;
             case NONE -> Format.PROTOBUF;
         };
     }
 
     /**
-     * Write a new zip file containing blocks, reads the batch of blocks from the HistoricalBlockFacility.
+     * Write a new zip file containing the input batch of blocks.
      *
-     * @param firstBlockNumber The first block number to write
+     * @return The size of the zip file created
      * @throws IOException If an error occurs writing the block
-     * @return The size of the zip file created, 0 if no blocks were written.
      */
-    long writeNewZipFile(long firstBlockNumber) throws IOException {
-        final long lastBlockNumber = firstBlockNumber + numberOfBlocksPerZipFile - 1;
-        // compute block path
-        final BlockPath firstBlockPath = computeBlockPath(config, firstBlockNumber);
+    long writeNewZipFile(List<BlockAccessor> batch) throws IOException {
+        // compute zip path
+        final BlockPath firstBlockPath =
+                computeBlockPath(config, batch.getFirst().blockNumber());
         // create directories
         Files.createDirectories(firstBlockPath.dirPath());
-        // create list for all block accessors, so we can delete files after we are done
-        final List<BlockAccessor> blockAccessors = IntStream.rangeClosed((int) firstBlockNumber, (int) lastBlockNumber)
-                .mapToObj(historicalBlockFacility::block)
-                .toList();
-        // create zip file path
-        try (ZipOutputStream zipOutputStream = new ZipOutputStream(new BufferedOutputStream(
-                Files.newOutputStream(
-                        firstBlockPath.zipFilePath(), StandardOpenOption.CREATE, StandardOpenOption.WRITE),
+        // create zip file
+        final Path zipFilePath = firstBlockPath.zipFilePath();
+        try (final ZipOutputStream zipOutputStream = new ZipOutputStream(new BufferedOutputStream(
+                Files.newOutputStream(zipFilePath, StandardOpenOption.CREATE, StandardOpenOption.WRITE),
                 1024 * 1204))) {
             // don't compress the zip file as files are already compressed
             zipOutputStream.setMethod(ZipOutputStream.STORED);
-            // todo should we not also set the level to Deflater.NO_COMPRESSION
-            for (long blockNumber = firstBlockNumber; blockNumber <= lastBlockNumber; blockNumber++) {
+            zipOutputStream.setLevel(Deflater.NO_COMPRESSION);
+            for (final BlockAccessor blockAccessor : batch) {
                 // compute block filename
-                // todo should we also not append the compression extension to the filename?
-                // todo I feel like the accessor should generally be getting us the block file name
-                //   what if the file is zstd compressed but the current runtime compression is none?
-                //   then the file name would be wrong? For now appending, maybe a slight cleanup is in order for this
-                //   logic.
-                final String blockFileName = BlockFile.blockFileName(blockNumber, config.compression());
-                // get block accessor
-                final BlockAccessor blockAccessor = blockAccessors.get((int) (blockNumber - firstBlockNumber));
+                final String blockFileName = BlockFile.blockFileName(blockAccessor.blockNumber(), config.compression());
                 // get the bytes to write, we have to do this as we need to know the size
                 final Bytes bytes = blockAccessor.blockBytes(format);
                 // calculate CRC-32 checksum
-                CRC32 crc = new CRC32();
+                final CRC32 crc = new CRC32();
                 crc.update(bytes.toByteArray());
                 // create zip entry
                 final ZipEntry zipEntry = new ZipEntry(blockFileName);
@@ -119,9 +99,10 @@ long writeNewZipFile(long firstBlockNumber) throws IOException {
                 zipOutputStream.closeEntry();
             }
         }
+        // if we have reached here, this means that the zip file was created
+        // successfully
         // return the size of the zip file created
-        // todo(1217): consider case when no zip is created and rerturn 0
-        return Files.size(firstBlockPath.zipFilePath());
+        return Files.size(zipFilePath);
     }
 
     /**