@Override
  public <E extends Exception> int forEach(final @NotNull Processor<E> processor) throws IOException, E {
    final long storageLength = pagedStorage.length();
    int currentId = offsetToId(recordsStartOffset());
    for (int recordNo = 0; ; recordNo++) {
      final long recordOffset = idToOffset(currentId);
      try (final PageUnsafe page = (PageUnsafe)pagedStorage.pageByOffset(recordOffset, /*forWrite: */ false)) {
        final int offsetOnPage = pagedStorage.toOffsetInPage(recordOffset);
        page.lockPageForRead();
        try {
          final ByteBuffer buffer = page.rawPageBuffer();
          final RecordLayout recordLayout = RecordLayout.recordLayout(buffer, offsetOnPage);
          final byte recordType = recordLayout.recordType();
          final int recordCapacity = recordLayout.capacity(buffer, offsetOnPage);
          switch (recordType) {
            case RecordLayout.RECORD_TYPE_ACTUAL, RecordLayout.RECORD_TYPE_MOVED -> {
              final int headerSize = recordLayout.headerSize();
              final boolean isActual = recordType == RecordLayout.RECORD_TYPE_ACTUAL;
              final int recordActualLength = isActual ? recordLayout.length(buffer, offsetOnPage) : -1;
              final ByteBuffer slice = isActual ?
                                       buffer.slice(offsetOnPage + headerSize, recordActualLength)
                                         .asReadOnlyBuffer()
                                         .order(buffer.order()) :
                                       buffer.slice(offsetOnPage + headerSize, 0)
                                         .asReadOnlyBuffer()
                                         .order(buffer.order());
              final boolean ok = processor.processRecord(currentId, recordCapacity, recordActualLength, slice);
              if (!ok) {
                return recordNo + 1;
              }
            }
            default -> {
              //just skip for now
            }
          }

          final long nextRecordOffset = nextRecordOffset(recordOffset, recordLayout, recordCapacity);
          if (nextRecordOffset >= storageLength) {
            return recordNo;
          }

          currentId = offsetToId(nextRecordOffset);
        }
        finally {
          page.unlockPageForRead();
        }
      }
    }
  }
@Override
  public <E extends Exception> int forEach(final @NotNull Processor<E> processor)