private boolean forEachRecord(@NotNull ChunkReader reader,
                                long untilOffset) throws IOException {
    int pageSize = storage.pageSize();
    for (long chunkOffsetInFile = FileHeader.HEADER_SIZE; chunkOffsetInFile < untilOffset; ) {

      MMappedFileStorage.Page page = storage.pageByOffset(chunkOffsetInFile);
      int chunkOffsetInPage = storage.toOffsetInPage(chunkOffsetInFile);
      ByteBuffer pageBuffer = page.rawPageBuffer();

      if (pageSize - chunkOffsetInPage < LogChunkImpl.HEADER_SIZE) {
        throw new CorruptedException(
          getClass().getSimpleName() + " corrupted: chunkOffsetInPage(=" + chunkOffsetInPage + ") less than " +
          "RECORD_HEADER(=" + LogChunkImpl.HEADER_SIZE + "b) left until " +
          "pageEnd(" + pageSize + ") -- all chunks must be 64b-aligned"
        );
      }

      int chunkHeader = LogChunkImpl.readHeader(pageBuffer, chunkOffsetInPage);
      if (!LogChunkImpl.isHeaderSet(chunkHeader)) {
        //the chunk wasn't even started to be written
        // -> can't read the following records since we don't know there they are
        return true;
      }

      int chunkLength = LogChunkImpl.unpackChunkLength(chunkHeader);

      if (LogChunkImpl.isDataChunk(chunkHeader)) {
        long chunkId = chunkOffsetToId(chunkOffsetInFile);

        if (!LogChunkImpl.isFitIntoPage(pageBuffer, chunkOffsetInPage, chunkLength)) {
          throw new CorruptedException("chunk[" + chunkId + "][@" + chunkOffsetInFile + "].chunkLength(=" + chunkLength + "): " +
                                       " is incorrect: page[0.." + pageBuffer.limit() + "]" +
                                       moreDiagnosticInfo(chunkOffsetInFile));
        }

        LogChunkImpl chunk = readChunkAt(pageBuffer, chunkOffsetInPage, chunkOffsetInFile);
        boolean shouldContinue = reader.read(chunk);
        if (!shouldContinue) {
          return false;
        }
      }
      else if (LogChunkImpl.isPaddingChunk(chunkHeader)) {
        //just skip it
      }
      else {
        //if header != 0 => it must be either padding, or (uncommitted?) data record:
        throw new IOException("header(=" + chunkHeader + "](@offset=" + chunkOffsetInFile + "): not a padding, nor a data chunk");
      }


      chunkOffsetInFile = nextChunkOffset(chunkOffsetInFile, chunkLength);
    }

    return true;
  }
private boolean forEachRecord(@NotNull ChunkReader reader,
                                long untilOffset)