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.
* Adjusted BlockFile to use a visitor for some file tree walks.

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/BlockFile.java

@@ -1,9 +1,17 @@
 // SPDX-License-Identifier: Apache-2.0
 package org.hiero.block.node.base;
 
+import static java.nio.file.FileVisitResult.CONTINUE;
+
 import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.lang.System.Logger;
+import java.lang.System.Logger.Level;
+import java.nio.file.FileVisitResult;
+import java.nio.file.FileVisitor;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.nio.file.attribute.BasicFileAttributes;
 import java.text.DecimalFormat;
 import java.text.NumberFormat;
 import java.util.Comparator;
@@ -21,12 +29,14 @@
 @SuppressWarnings("unused")
 public final class BlockFile {
     /** The logger for this class. */
-    private static final System.Logger LOGGER = System.getLogger(BlockFile.class.getName());
+    private static final Logger LOGGER = System.getLogger(BlockFile.class.getName());
     /** The format for block numbers in file names */
     private static final NumberFormat BLOCK_NUMBER_FORMAT = new DecimalFormat("0000000000000000000");
     /** The extension for compressed block files */
     public static final String BLOCK_FILE_EXTENSION = ".blk";
 
+    private static final int MAX_FILE_SEARCH_DEPTH = 20;
+
     /**
      * The block number format for use in file name. For example block 123 becomes "000000000000000123".
      *
@@ -134,7 +144,7 @@ public static long nestedDirectoriesMinBlockNumber(Path basePath, CompressionTyp
                     break;
                 }
             } catch (Exception e) {
-                LOGGER.log(System.Logger.Level.ERROR, "Error reading directory: " + lowestPath, e);
+                LOGGER.log(Level.ERROR, "Error reading directory: " + lowestPath, e);
                 // handle exception
                 break;
             }
@@ -178,7 +188,7 @@ public static long nestedDirectoriesMaxBlockNumber(Path basePath, CompressionTyp
                     break;
                 }
             } catch (Exception e) {
-                LOGGER.log(System.Logger.Level.ERROR, "Error reading directory: " + highestPath, e);
+                LOGGER.log(Level.ERROR, "Error reading directory: " + highestPath, e);
                 // handle exception
                 break;
             }
@@ -195,16 +205,18 @@ public static long nestedDirectoriesMaxBlockNumber(Path basePath, CompressionTyp
      * @return the minimum block number, or -1 if no block files are found
      */
     public static Set<Long> nestedDirectoriesAllBlockNumbers(Path basePath, CompressionType compressionType) {
-        final Set<Long> blockNumbers = new HashSet<>();
-        final String fullExtension = BLOCK_FILE_EXTENSION + compressionType.extension();
-        try (var stream = Files.walk(basePath)) {
-            stream.filter(Files::isRegularFile)
-                    .filter(path -> path.getFileName().toString().endsWith(fullExtension))
-                    .forEach(blockFile -> blockNumbers.add(blockNumberFromFile(blockFile)));
+        try {
+            final Set<Long> blockNumbers = new HashSet<>();
+            final String fullExtension = BLOCK_FILE_EXTENSION + compressionType.extension();
+            Files.walkFileTree(
+                    basePath,
+                    Set.of(),
+                    MAX_FILE_SEARCH_DEPTH,
+                    new BlockNumberCollectionVisitor(blockNumbers, fullExtension));
+            return blockNumbers;
         } catch (IOException e) {
-            throw new RuntimeException(e);
+            throw new UncheckedIOException(e);
         }
-        return blockNumbers;
     }
 
     /**
@@ -213,21 +225,60 @@ public static Set<Long> nestedDirectoriesAllBlockNumbers(Path basePath, Compress
      * extension.
      *
      * @param file the path for file to extract the block number from
-     * @return the block number
+     * @return the block number, or `-1` if the file does not match the expected format
+     *     or the file base name does not parse as a `long` value.
      */
     public static long blockNumberFromFile(final Path file) {
-        return blockNumberFromFile(file.getFileName().toString());
+        try {
+            if (file != null && file.getFileName() != null) {
+                final String fileName = file.getFileName().toString();
+                if (fileName != null && fileName.indexOf('.') > 0) {
+                    return Long.parseLong(fileName.substring(0, fileName.indexOf('.')));
+                } else {
+                    return -1L;
+                }
+            } else {
+                return -1L;
+            }
+        } catch (NumberFormatException e) {
+            return -1L;
+        }
     }
 
-    /**
-     * Extracts the block number from a file name. The file name is expected to be in the format
-     * {@code "0000000000000000000.blk.xyz"} where the block number is the first 19 digits and the rest is the file
-     * extension.
-     *
-     * @param fileName the file name to extract the block number from
-     * @return the block number
-     */
-    public static long blockNumberFromFile(final String fileName) {
-        return Long.parseLong(fileName.substring(0, fileName.indexOf('.')));
+    private static class BlockNumberCollectionVisitor implements FileVisitor<Path> {
+        private final String blockFileExtension;
+        private final Set<Long> blockNumbersFound;
+
+        public BlockNumberCollectionVisitor(final Set<Long> blockNumbers, final String fullExtension) {
+            blockFileExtension = fullExtension;
+            blockNumbersFound = blockNumbers;
+        }
+
+        @Override
+        public FileVisitResult preVisitDirectory(final Path dir, final BasicFileAttributes attrs) throws IOException {
+            return CONTINUE;
+        }
+
+        @Override
+        public FileVisitResult visitFile(final Path file, final BasicFileAttributes attrs) throws IOException {
+            if (file.endsWith(blockFileExtension)) {
+                blockNumbersFound.add(BlockFile.blockNumberFromFile(file));
+            }
+            return CONTINUE;
+        }
+
+        @Override
+        public FileVisitResult visitFileFailed(final Path file, final IOException exc) throws IOException {
+            throw exc;
+        }
+
+        @Override
+        public FileVisitResult postVisitDirectory(final Path dir, final IOException exc) throws IOException {
+            if (exc != null) {
+                return CONTINUE;
+            } else {
+                throw exc;
+            }
+        }
     }
 }

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/base/src/test/java/org/hiero/block/node/base/BlockFileTest.java

@@ -1,15 +1,25 @@
 // SPDX-License-Identifier: Apache-2.0
 package org.hiero.block.node.base;
 
-import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.fail;
 
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.Arrays;
 import java.util.Set;
+import java.util.stream.Stream;
+import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.aggregator.ArgumentsAccessor;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 
 /**
  * Unit tests for the BlockFile class.
@@ -33,10 +43,41 @@ void testStandaloneBlockFilePath() {
         assertEquals(expectedPath, BlockFile.standaloneBlockFilePath(basePath, 123, CompressionType.ZSTD));
     }
 
-    @Test
-    void testBlockNumberFromFile() {
-        Path filePath = Paths.get("0000000000000000123.blk.zstd");
-        assertEquals(123, BlockFile.blockNumberFromFile(filePath));
+    @ParameterizedTest(name = "{0}")
+    @MethodSource("org.hiero.block.node.base.BlockFileTest#blockNumbersAndFilePaths")
+    @DisplayName("Block Number From File")
+    void testBlockNumberFromFile(final ArgumentsAccessor args) {
+        final long expectedNumber = args.getLong(1);
+        final Path file = args.get(2, Path.class);
+        assertEquals(expectedNumber, BlockFile.blockNumberFromFile(file));
+    }
+
+    /**
+     * Method source for test cases to verify `blockNumberFromFile`.
+     */
+    private static Stream<Arguments> blockNumbersAndFilePaths() {
+        final String thirtySixZeros = "000000000000000000000000000000000000";
+        return Stream.of(
+                Arguments.of("Simple 'happy' test", 123L, Path.of("0000000000000000123.blk.zstd")),
+                Arguments.of("a large block number", 1827329783712391L, Path.of("0001827329783712391.blk.zstd")),
+                Arguments.of("max-long block number", Long.MAX_VALUE, Path.of(Long.MAX_VALUE + ".blk.zstd")),
+                Arguments.of("number 0", 0L, Path.of("0.blk.zstd")),
+                Arguments.of("number 0 with extra 0's prefix", 0L, Path.of(thirtySixZeros + "0000.blk.zstd")),
+                Arguments.of("number 1984 with extra 0's prefix", 1984L, Path.of(thirtySixZeros + "1984.blk.zstd")),
+                Arguments.of(
+                        "directory that looks like a block file",
+                        10L,
+                        Path.of("/000/000/000/010.blk.zip/nothing").getParent()),
+                Arguments.of("Wrong extension returns -1", -1L, Path.of("marker.file")),
+                Arguments.of("No extension returns -1", -1L, Path.of("file-without-extension")),
+                Arguments.of("Filename is not a number returns -1", -1L, Path.of("copy-of-error.blk.zstd")),
+                Arguments.of("Filename is out of range returns -1", -1L, Path.of(Long.MAX_VALUE + "9182.blk.zstd")),
+                Arguments.of("Directory without an extension returns -1", -1L, Path.of("/000/000/000/010")),
+                Arguments.of(
+                        "root path (getFileName is null) returns -1",
+                        -1L,
+                        Path.of("/000/000/000/010/").getRoot()),
+                Arguments.of("null path returns -1", -1L, null));
     }
 
     /**