from __future__ import annotations
from dataclasses import dataclass
from typing import Iterator, List, Optional
-from warnings import warn
import argparse
import contextlib
import hashlib
def _do_verify(conn, disk_id: int, target_ranges: List[range], disk_file: io.BufferedIOBase, read_size: int, read_tries: int):
- requested_blocks = {
- block
- for r in target_ranges
- for block in r
- }
pg_target_ranges = [NumericRange(r.start, r.stop) for r in target_ranges]
# transaction is required for named cursors in worklist generation
key = _run_sort_key,
)
- covered_blocks = set()
passes = []
fails = []
for run in worklist:
else:
fails.extend(run.block_ranges)
- covered_blocks.update(
- block
- for block_range in run.block_ranges
- for block in block_range
- )
-
- missing = requested_blocks - covered_blocks
- if missing:
- warn(f"unable to locate blocks: {len(missing)} in the range {min(missing)} to {max(missing)}")
-
def clean_up(ranges):
out = []
for r in sorted(ranges, key = lambda r: r.start):