chore: BlockAccessors Hardening & Improved Error Handling

* BlockAccessors are now Autocloseable
* BlockAccessors now create hard links for the duration of their
  lifespan
  * This allows us to ensure data is accessible for the duration of the
    accessor's life
* Tests are updated and migrated
* Deleted redundant configuration tests
* Added tests:
  * Tests for subsequent reads to ensure that closing an accessor does
    not affect the data or the ability for a new accessor to access it
* Added a recursive deletion visitor
* Introduced a BlockAccessorBatch
* Added reuse of `ZipFileSystem` instances to cut down on
  repeated reread of large zip files and dictionaries.
* Adjusted block accessor close semantics to avoid double-close.

Signed-off-by: Atanas Atanasov <a.v.atanasov98@gmail.com>
Signed-off-by: Joseph S. <121976561+jsync-swirlds@users.noreply.github.com>

Author: ata-nas

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

@@ -4,7 +4,14 @@
 import com.hedera.hapi.block.stream.Block;
 import com.hedera.pbj.runtime.ParseException;
 import com.hedera.pbj.runtime.io.buffer.Bytes;
+import edu.umd.cs.findbugs.annotations.NonNull;
+import edu.umd.cs.findbugs.annotations.Nullable;
 import java.io.IOException;
+import java.nio.file.FileVisitResult;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.attribute.BasicFileAttributes;
 import java.util.zip.GZIPInputStream;
 import org.hiero.block.internal.BlockUnparsed;
 import org.hiero.block.node.app.fixtures.TestUtils;
@@ -112,4 +119,31 @@ public long getBlockNumber() {
             return blockNumber;
         }
     }
+
+    /**
+     * A simple file visitor to recursively delete files and directories up to
+     * the provided root.
+     */
+    private static class RecursiveFileDeleteVisitor extends SimpleFileVisitor<Path> {
+        @Override
+        @NonNull
+        public FileVisitResult visitFile(@NonNull final Path file, @NonNull final BasicFileAttributes attrs)
+                throws IOException {
+            Files.delete(file);
+            return FileVisitResult.CONTINUE;
+        }
+
+        @Override
+        @NonNull
+        public FileVisitResult postVisitDirectory(@NonNull final Path dir, @Nullable final IOException e)
+                throws IOException {
+            if (e == null) {
+                Files.delete(dir);
+                return FileVisitResult.CONTINUE;
+            } else {
+                // directory iteration failed
+                throw e;
+            }
+        }
+    }
 }

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

@@ -18,6 +18,8 @@ public final class InMemoryBlockAccessor implements BlockAccessor {
     private final Block block;
     /** The block number of the accessible block */
     private final long blockNumber;
+    /** simple flag to track calls to close */
+    private boolean isClosed = false;
 
     /**
      * Constructor. Enforces preconditions on the input block items.
@@ -85,4 +87,14 @@ public Bytes blockBytes(final Format format) {
     private Bytes zstdCompressBytes(final Bytes bytes) {
         return Bytes.wrap(Zstd.compress(bytes.toByteArray()));
     }
+
+    @Override
+    public void close() {
+        isClosed = true;
+    }
+
+    @Override
+    public boolean isClosed() {
+        return isClosed;
+    }
 }

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

@@ -9,6 +9,7 @@
 public final class MinimalBlockAccessor implements BlockAccessor {
     private final long blockNumber;
     private final Block block;
+    private boolean isClosed = false;
 
     public MinimalBlockAccessor(final long blockNumber, final Block block) {
         this.blockNumber = blockNumber;
@@ -32,4 +33,14 @@ public Bytes blockBytes(final Format format) {
     private Bytes zstdCompressBytes(final Bytes bytes) {
         return Bytes.wrap(Zstd.compress(bytes.toByteArray()));
     }
+
+    @Override
+    public void close() {
+        isClosed = true;
+    }
+
+    @Override
+    public boolean isClosed() {
+        return isClosed;
+    }
 }

block-node/base/src/main/java/org/hiero/block/node/base/tar/TaredBlockIterator.java

@@ -1,12 +1,14 @@
 // SPDX-License-Identifier: Apache-2.0
 package org.hiero.block.node.base.tar;
 
+import edu.umd.cs.findbugs.annotations.NonNull;
 import java.nio.charset.StandardCharsets;
 import java.text.DecimalFormat;
 import java.text.NumberFormat;
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.NoSuchElementException;
+import java.util.Objects;
 import org.hiero.block.node.spi.historicalblocks.BlockAccessor;
 import org.hiero.block.node.spi.historicalblocks.BlockAccessor.Format;
 
@@ -59,13 +61,14 @@ enum State {
      * @param format the format of the block data files written to the tar file
      * @param blockIterator the iterator over the blocks to be written into the tar file
      */
-    public TaredBlockIterator(Format format, Iterator<BlockAccessor> blockIterator) {
-        this.format = format;
+    public TaredBlockIterator(@NonNull final Format format, @NonNull final Iterator<BlockAccessor> blockIterator) {
+        this.format = Objects.requireNonNull(format);
         this.extension = switch (format) {
             case Format.JSON -> ".json";
             case Format.PROTOBUF -> ".blk";
-            case Format.ZSTD_PROTOBUF -> ".blk.zstd";};
-        this.blockIterator = blockIterator;
+            case Format.ZSTD_PROTOBUF -> ".blk.zstd";
+        };
+        this.blockIterator = Objects.requireNonNull(blockIterator);
         this.currentBuffer = new byte[CHUNK_SIZE];
         this.bufferPosition = 0;
     }
@@ -98,9 +101,17 @@ public byte[] next() {
                 case NEXT_FILE:
                     if (blockIterator.hasNext()) {
                         // get the next block from the iterator
-                        final BlockAccessor currentBlock = blockIterator.next();
-                        currentBlockBytes = currentBlock.blockBytes(format).toByteArray();
-                        currentBlockName = BLOCK_NUMBER_FORMAT.format(currentBlock.blockNumber()) + extension;
+                        final BlockAccessor currentBlockAccessor = blockIterator.next();
+                        currentBlockName = BLOCK_NUMBER_FORMAT.format(currentBlockAccessor.blockNumber()) + extension;
+                        currentBlockBytes =
+                                currentBlockAccessor.blockBytes(format).toByteArray();
+                        // close the accessor as we are done with it and we need
+                        // to free resources, but only if it isn't already closed.
+                        // @todo replace Iterator with BlockAccessorBatch for this
+                        //     class so we don't need to call close here.
+                        if (!currentBlockAccessor.isClosed()) {
+                            currentBlockAccessor.close();
+                        }
                         filePosition = 0;
                         state = State.WRITE_HEADER;
                     } else {

block-node/block-access/src/main/java/org/hiero/block/node/access/service/BlockAccessServicePlugin.java

@@ -5,7 +5,6 @@
 import static java.lang.System.Logger.Level.INFO;
 import static java.lang.System.Logger.Level.TRACE;
 
-import com.hedera.hapi.block.stream.Block;
 import com.swirlds.metrics.api.Counter;
 import org.hiero.block.api.BlockAccessServiceInterface;
 import org.hiero.block.api.BlockRequest;
@@ -14,6 +13,7 @@
 import org.hiero.block.node.spi.BlockNodeContext;
 import org.hiero.block.node.spi.BlockNodePlugin;
 import org.hiero.block.node.spi.ServiceBuilder;
+import org.hiero.block.node.spi.historicalblocks.BlockAccessor;
 import org.hiero.block.node.spi.historicalblocks.HistoricalBlockFacility;
 
 /**
@@ -44,12 +44,10 @@ public class BlockAccessServicePlugin implements BlockNodePlugin, BlockAccessSer
      */
     public BlockResponse getBlock(BlockRequest request) {
         requestCounter.increment();
-
         try {
             long blockNumberToRetrieve;
             // The block number and retrieve latest are mutually exclusive in
             // the proto definition, so no need to check for that here.
-
             // if retrieveLatest is set, or the request is for the largest possible block number, get the latest block.
             if (request.hasBlockNumber() && request.blockNumber() >= 0) {
                 blockNumberToRetrieve = request.blockNumber();
@@ -63,7 +61,6 @@ public BlockResponse getBlock(BlockRequest request) {
                 LOGGER.log(INFO, "Invalid request, 'retrieve_latest' or a valid 'block number' is required.");
                 return new BlockResponse(Code.INVALID_REQUEST, null);
             }
-
             // Check if block is within the available range
             if (!blockProvider.availableBlocks().contains(blockNumberToRetrieve)) {
                 long lowestBlockNumber = blockProvider.availableBlocks().min();
@@ -77,14 +74,21 @@ public BlockResponse getBlock(BlockRequest request) {
                 responseCounterNotAvailable.increment();
                 return new BlockResponse(Code.NOT_AVAILABLE, null);
             }
-
             // Retrieve the block
-            Block block = blockProvider.block(blockNumberToRetrieve).block();
-            responseCounterSuccess.increment();
-            return new BlockResponse(Code.SUCCESS, block);
-
-        } catch (RuntimeException e) {
-            String message = "Failed to retrieve block number %d.".formatted(request.blockNumber());
+            try (final BlockAccessor accessor = blockProvider.block(blockNumberToRetrieve)) {
+                if (accessor != null) {
+                    // even though we have checked for the existence of the block
+                    // it may not be available anymore, so we have to ensure the accessor
+                    // still exists
+                    final BlockResponse response = new BlockResponse(Code.SUCCESS, accessor.block());
+                    responseCounterSuccess.increment();
+                    return response;
+                } else {
+                    return new BlockResponse(Code.NOT_FOUND, null);
+                }
+            }
+        } catch (final RuntimeException e) {
+            final String message = "Failed to retrieve block number %d.".formatted(request.blockNumber());
             LOGGER.log(ERROR, message, e);
             responseCounterNotFound.increment(); // Should this be a failure counter?
             return new BlockResponse(Code.NOT_FOUND, null);