leaderless_log_reader.py

from __future__ import annotations

import os
import time
from dataclasses import dataclass
from typing import Any, Iterator, Protocol

import oxia
import oxia.ex

from .leaderless_log_common import (
    ControlRecord,
    EntryType,
    S3ClientConfig,
    TopicPartition,
    TopicPartitionPaths,
    WalIndexEntry,
    decode_batch_layout,
    json_loads,
    parse_s3_uri,
    resolve_s3_client_config,
)
from .leaderless_log_metrics import LeaderlessLogMetrics


class LeaderlessLogReaderError(Exception):
    pass


class OffsetOutOfRange(LeaderlessLogReaderError):
    pass


class InconsistentReadState(LeaderlessLogReaderError):
    pass


class PartitionNotInitialized(LeaderlessLogReaderError):
    pass


class UnsupportedEntryType(LeaderlessLogReaderError):
    pass


class UnsupportedEncoding(LeaderlessLogReaderError):
    pass


class MalformedWalBlob(LeaderlessLogReaderError):
    pass


class BlobNotFound(LeaderlessLogReaderError):
    pass


class BlobReader(Protocol):
    def get_bytes(self, uri: str) -> bytes: ...

    def get_range(self, uri: str, byte_offset: int, byte_length: int) -> bytes: ...


class CompactedEntryReader(Protocol):
    def read_offset(self, entry: "ResolvedEntry", offset: int) -> bytes: ...


class S3BlobReader:
    def __init__(
        self,
        *,
        config: S3ClientConfig | None = None,
        client: Any | None = None,
        endpoint_url: str | None = None,
        region_name: str | None = None,
        access_key_id: str | None = None,
        secret_access_key: str | None = None,
        session_token: str | None = None,
        metrics: LeaderlessLogMetrics | None = None,
    ) -> None:
        resolved = resolve_s3_client_config(
            config,
            endpoint_url=endpoint_url,
            region_name=region_name,
            access_key_id=access_key_id,
            secret_access_key=secret_access_key,
            session_token=session_token,
        )
        self._client = client or self._build_client(resolved)
        self._metrics = metrics

    @staticmethod
    def _build_client(config: S3ClientConfig) -> Any:
        try:
            import boto3
            from botocore.config import Config
        except ImportError as exc:  # pragma: no cover - depends on runtime env
            raise LeaderlessLogReaderError(
                "boto3 is required for the default S3 blob reader"
            ) from exc

        return boto3.client(
            "s3",
            endpoint_url=config.endpoint_url,
            region_name=config.region_name,
            aws_access_key_id=config.access_key_id,
            aws_secret_access_key=config.secret_access_key,
            aws_session_token=config.session_token,
            config=Config(s3={"addressing_style": "path"}),
        )

    def get_bytes(self, uri: str) -> bytes:
        bucket, key = parse_s3_uri(uri)
        started = time.perf_counter()
        result = "success"
        try:
            response = self._client.get_object(Bucket=bucket, Key=key)
        except Exception as exc:
            response = getattr(exc, "response", None)
            if isinstance(response, dict):
                code = response.get("Error", {}).get("Code")
                if code in {"404", "NoSuchBucket", "NoSuchKey"}:
                    result = "not_found"
                    raise BlobNotFound(uri) from exc
            result = "error"
            raise
        try:
            return self._read_body(response)
        finally:
            if self._metrics is not None:
                self._metrics.observe_s3_operation(
                    operation="get",
                    result=result,
                    duration_seconds=time.perf_counter() - started,
                )

    def get_range(self, uri: str, byte_offset: int, byte_length: int) -> bytes:
        bucket, key = parse_s3_uri(uri)
        started = time.perf_counter()
        result = "success"
        try:
            response = self._client.get_object(
                Bucket=bucket,
                Key=key,
                Range=f"bytes={byte_offset}-{byte_offset + byte_length - 1}",
            )
        except Exception as exc:
            response = getattr(exc, "response", None)
            if isinstance(response, dict):
                code = response.get("Error", {}).get("Code")
                if code in {"404", "NoSuchBucket", "NoSuchKey"}:
                    result = "not_found"
                    raise BlobNotFound(uri) from exc
            result = "error"
            raise
        try:
            return self._read_body(response)
        finally:
            if self._metrics is not None:
                self._metrics.observe_s3_operation(
                    operation="range_get",
                    result=result,
                    duration_seconds=time.perf_counter() - started,
                )

    @staticmethod
    def _read_body(response: dict[str, Any]) -> bytes:
        body = response["Body"]
        try:
            return body.read()
        finally:
            close = getattr(body, "close", None)
            if callable(close):
                close()


@dataclass(frozen=True)
class ResolvedEntry:
    start_offset: int
    end_offset: int
    msg_count: int
    entry_type: EntryType
    data_key: str
    encoding: str
    created_at_ms: int
    source: str
    byte_offset: int | None = None
    byte_length: int | None = None

    def covers(self, offset: int) -> bool:
        return self.start_offset <= offset <= self.end_offset


@dataclass(frozen=True)
class ReadResult:
    topic: str
    partition: int
    offset: int
    payload: bytes
    entry_type: EntryType
    start_offset: int
    end_offset: int
    data_key: str
    encoding: str


class LeaderlessLogReader:
    def __init__(
        self,
        service_address: str,
        namespace: str = "default",
        root_prefix: str = "llog",
        session_timeout_ms: int = 30_000,
        client_identifier: str | None = None,
        blob_reader: BlobReader | None = None,
        compacted_reader: CompactedEntryReader | None = None,
        s3_config: S3ClientConfig | None = None,
        s3_endpoint_url: str | None = None,
        s3_region_name: str | None = None,
        s3_access_key_id: str | None = None,
        s3_secret_access_key: str | None = None,
        s3_session_token: str | None = None,
        metrics: LeaderlessLogMetrics | None = None,
    ) -> None:
        self._client = oxia.Client(
            service_address,
            namespace=namespace,
            session_timeout_ms=session_timeout_ms,
            client_identifier=client_identifier,
        )
        self._root_prefix = root_prefix.rstrip("/")
        self._metrics = metrics
        if blob_reader is not None:
            self._blob_reader = blob_reader
        elif s3_config is not None:
            if any(
                value is not None
                for value in (
                    s3_endpoint_url,
                    s3_region_name,
                    s3_access_key_id,
                    s3_secret_access_key,
                    s3_session_token,
                )
            ):
                self._blob_reader = S3BlobReader(
                    config=s3_config,
                    endpoint_url=s3_endpoint_url,
                    region_name=s3_region_name,
                    access_key_id=s3_access_key_id,
                    secret_access_key=s3_secret_access_key,
                    session_token=s3_session_token,
                    metrics=metrics,
                )
            else:
                self._blob_reader = S3BlobReader(config=s3_config, metrics=metrics)
        else:
            self._blob_reader = S3BlobReader(
                endpoint_url=s3_endpoint_url,
                region_name=s3_region_name,
                access_key_id=s3_access_key_id,
                secret_access_key=s3_secret_access_key,
                session_token=s3_session_token,
                metrics=metrics,
            )
        self._compacted_reader = compacted_reader

    @classmethod
    def from_env(
        cls,
        *,
        service_address: str | None = None,
        namespace: str | None = None,
        root_prefix: str | None = None,
        session_timeout_ms: int | None = None,
        client_identifier: str | None = None,
        blob_reader: BlobReader | None = None,
        compacted_reader: CompactedEntryReader | None = None,
        metrics: LeaderlessLogMetrics | None = None,
    ) -> "LeaderlessLogReader":
        resolved_service_address = service_address or cls._require_env(
            "LLOG_OXIA_SERVICE_ADDRESS"
        )
        resolved_namespace = namespace or os.environ.get("LLOG_OXIA_NAMESPACE", "default")
        resolved_root_prefix = root_prefix or os.environ.get("LLOG_ROOT_PREFIX", "llog")
        resolved_session_timeout_ms = (
            session_timeout_ms
            if session_timeout_ms is not None
            else int(os.environ.get("LLOG_OXIA_SESSION_TIMEOUT_MS", "30000"))
        )
        resolved_client_identifier = client_identifier or os.environ.get(
            "LLOG_CLIENT_IDENTIFIER"
        )

        if blob_reader is not None:
            return cls(
                resolved_service_address,
                namespace=resolved_namespace,
                root_prefix=resolved_root_prefix,
                session_timeout_ms=resolved_session_timeout_ms,
                client_identifier=resolved_client_identifier,
                blob_reader=blob_reader,
                compacted_reader=compacted_reader,
                metrics=metrics,
            )

        resolved_s3_endpoint_url = cls._optional_env("LLOG_S3_ENDPOINT_URL")
        resolved_s3_region_name = cls._optional_env(
            "LLOG_S3_REGION_NAME",
            "AWS_REGION",
            "AWS_DEFAULT_REGION",
        ) or "us-east-1"
        resolved_s3_access_key_id = cls._optional_env(
            "LLOG_S3_ACCESS_KEY_ID",
            "AWS_ACCESS_KEY_ID",
        )
        resolved_s3_secret_access_key = cls._optional_env(
            "LLOG_S3_SECRET_ACCESS_KEY",
            "AWS_SECRET_ACCESS_KEY",
        )
        resolved_s3_session_token = cls._optional_env(
            "LLOG_S3_SESSION_TOKEN",
            "AWS_SESSION_TOKEN",
        )
        resolved_s3_config = S3ClientConfig(
            endpoint_url=resolved_s3_endpoint_url,
            region_name=resolved_s3_region_name,
            access_key_id=resolved_s3_access_key_id,
            secret_access_key=resolved_s3_secret_access_key,
            session_token=resolved_s3_session_token,
        )

        return cls(
            resolved_service_address,
            namespace=resolved_namespace,
            root_prefix=resolved_root_prefix,
            session_timeout_ms=resolved_session_timeout_ms,
            client_identifier=resolved_client_identifier,
            compacted_reader=compacted_reader,
            s3_config=resolved_s3_config,
            metrics=metrics,
        )

    def __enter__(self) -> "LeaderlessLogReader":
        return self

    def __exit__(self, exc_type, exc, tb) -> None:
        self.close()

    def close(self) -> None:
        self._client.close()

    def get_high_watermark(self, topic: str, partition: int) -> int:
        tp = self._tp(topic, partition)
        control = self._get_control(tp)
        return control.sequence_counter - 1

    def get_control_record(self, topic: str, partition: int) -> ControlRecord:
        return self._get_control(self._tp(topic, partition))

    def iter_index_entries(
        self,
        topic: str,
        partition: int,
        start_offset: int,
        stop_before: int,
    ) -> Iterator[ResolvedEntry]:
        return self._iter_index_entries(self._tp(topic, partition), start_offset, stop_before)

    @staticmethod
    def get_pending_cover(
        control: ControlRecord,
        offset: int,
    ) -> ResolvedEntry | None:
        return LeaderlessLogReader._get_pending_cover(control, offset)

    def read_blob(self, entry: ResolvedEntry) -> bytes:
        return self._read_blob(entry)

    def read_blob_full(self, data_key: str) -> bytes:
        try:
            return self._blob_reader.get_bytes(data_key)
        except KeyError as exc:
            raise BlobNotFound(data_key) from exc

    def decode_record(
        self,
        entry: ResolvedEntry,
        offset: int,
        blob: bytes,
    ) -> bytes:
        return self._decode_entry_message(entry, offset, self._resolve_entry_blob(entry, blob))

    def read_offset(self, topic: str, partition: int, offset: int) -> ReadResult:
        # Read path:
        #   1. Validate the offset is below the high watermark
        #   2. Resolve the entry: check the index (ceiling lookup), fall back to pending
        #   3. Fetch the WAL blob from S3 and decode the target message
        tp = self._tp(topic, partition)
        if offset < 1:
            raise ValueError("offset must be >= 1")

        # 1. VALIDATE: read the control record to get the high watermark.
        control = self._get_control(tp)
        if offset >= control.sequence_counter:
            raise OffsetOutOfRange(
                f"offset {offset} is >= high watermark {control.sequence_counter - 1}"
            )

        # 2. RESOLVE: find the index entry covering this offset. If the index
        #    doesn't have it yet, check the control record's pending field —
        #    the most recent append may still be in-flight (written to S3 but
        #    not yet indexed in Oxia).
        entry = self._get_index_ceiling(tp, offset)
        if entry is None or not entry.covers(offset):
            entry = self._get_pending_cover(control, offset)
        if entry is None or not entry.covers(offset):
            raise InconsistentReadState(
                f"no readable entry covers offset {offset} for {tp.topic}[{tp.partition}]"
            )

        # 3. FETCH & DECODE: download the WAL blob (or byte range) from S3
        #    and extract the individual message at the requested offset.
        payload = self._decode_entry_message(
            entry,
            offset,
            self._resolve_entry_blob(entry, self._read_blob(entry)),
        )
        return ReadResult(
            topic=tp.topic,
            partition=tp.partition,
            offset=offset,
            payload=payload,
            entry_type=entry.entry_type,
            start_offset=entry.start_offset,
            end_offset=entry.end_offset,
            data_key=entry.data_key,
            encoding=entry.encoding,
        )

    def read_range(
        self,
        topic: str,
        partition: int,
        start_offset: int,
        limit: int,
    ) -> list[ReadResult]:
        if limit <= 0:
            raise ValueError("limit must be > 0")
        return list(
            self.iter_from(
                topic,
                partition,
                start_offset=start_offset,
                max_messages=limit,
            )
        )

    def iter_from(
        self,
        topic: str,
        partition: int,
        start_offset: int = 1,
        stop_before: int | None = None,
        max_messages: int | None = None,
    ) -> Iterator[ReadResult]:
        tp = self._tp(topic, partition)
        if start_offset < 1:
            raise ValueError("start_offset must be >= 1")
        if stop_before is not None and stop_before < 1:
            raise ValueError("stop_before must be >= 1")
        if max_messages is not None and max_messages <= 0:
            raise ValueError("max_messages must be > 0")

        control = self._get_control(tp)
        snapshot_stop_before = control.sequence_counter
        if stop_before is not None:
            snapshot_stop_before = min(snapshot_stop_before, stop_before)
        if start_offset >= snapshot_stop_before:
            return iter(())

        def generator() -> Iterator[ReadResult]:
            yielded = 0
            next_offset = start_offset

            for entry in self._iter_index_entries(tp, start_offset, snapshot_stop_before):
                if entry.end_offset < next_offset:
                    continue
                if entry.start_offset > next_offset:
                    raise InconsistentReadState(
                        f"gap before offset {entry.start_offset} in {tp.topic}[{tp.partition}]"
                    )
                for result in self._iter_messages_from_entry(
                    tp,
                    entry,
                    next_offset,
                    snapshot_stop_before,
                ):
                    yield result
                    yielded += 1
                    next_offset = result.offset + 1
                    if max_messages is not None and yielded >= max_messages:
                        return
                if next_offset >= snapshot_stop_before:
                    return

            # The most recent append may still be pending (reserved in the
            # control record but not yet written to the index). Check it so
            # that reads immediately after a write see the new data.
            pending = self._get_pending_cover(control, next_offset)
            if pending is not None:
                for result in self._iter_messages_from_entry(
                    tp,
                    pending,
                    next_offset,
                    snapshot_stop_before,
                ):
                    yield result
                    yielded += 1
                    next_offset = result.offset + 1
                    if max_messages is not None and yielded >= max_messages:
                        return

            if next_offset < snapshot_stop_before:
                raise InconsistentReadState(
                    f"missing readable tail starting at offset {next_offset} in {tp.topic}[{tp.partition}]"
                )

        return generator()

    def _get_control(self, tp: TopicPartition) -> ControlRecord:
        try:
            _, value, _ = self._oxia_get(
                self._paths(tp).control_key,
                partition_key=self._oxia_partition_key(tp),
            )
        except oxia.ex.KeyNotFound as exc:
            raise PartitionNotInitialized(
                f"log partition {tp.topic}[{tp.partition}] is not initialized"
            ) from exc
        return self._json_loads_control(value)

    def _get_index_ceiling(self, tp: TopicPartition, offset: int) -> ResolvedEntry | None:
        # Index keys are keyed by end_offset. A CEILING lookup on the
        # requested offset finds the first entry whose end_offset >= offset,
        # i.e. the entry whose range may cover it.
        try:
            key, value, _ = self._oxia_get(
                self._index_lookup_key(tp, offset),
                partition_key=self._oxia_partition_key(tp),
                comparison_type=oxia.ComparisonType.CEILING,
            )
        except oxia.ex.KeyNotFound:
            return None

        paths = self._paths(tp)
        if not key.startswith(f"{paths.index_prefix}/"):
            return None

        entry = WalIndexEntry.from_dict(
            self._json_loads(value),
            error_type=InconsistentReadState,
        )
        end_offset = self._parse_offset_from_index_key(key)
        start_offset = end_offset - entry.msg_count + 1
        if start_offset <= 0:
            raise InconsistentReadState(f"invalid sparse index key {key}")
        return ResolvedEntry(
            start_offset=start_offset,
            end_offset=end_offset,
            msg_count=entry.msg_count,
            entry_type=entry.entry_type,
            data_key=entry.data_key,
            encoding=entry.encoding,
            created_at_ms=entry.created_at_ms,
            source="index",
            byte_offset=entry.byte_offset,
            byte_length=entry.byte_length,
        )

    @staticmethod
    def _get_pending_cover(control: ControlRecord, offset: int) -> ResolvedEntry | None:
        pending = control.pending
        if pending is None or not (pending.start_offset <= offset <= pending.end_offset):
            return None
        return ResolvedEntry(
            start_offset=pending.start_offset,
            end_offset=pending.end_offset,
            msg_count=pending.msg_count,
            entry_type=pending.entry_type,
            data_key=pending.data_key,
            encoding=pending.encoding,
            created_at_ms=pending.created_at_ms,
            source="pending",
            byte_offset=pending.byte_offset,
            byte_length=pending.byte_length,
        )

    def _iter_index_entries(
        self,
        tp: TopicPartition,
        start_offset: int,
        stop_before: int,
    ) -> Iterator[ResolvedEntry]:
        for key, value, _ in self._oxia_range_scan(
            self._index_lookup_key(tp, start_offset),
            self._index_scan_end_key(tp),
            partition_key=self._oxia_partition_key(tp),
        ):
            entry = WalIndexEntry.from_dict(
                self._json_loads(value),
                error_type=InconsistentReadState,
            )
            end_offset = self._parse_offset_from_index_key(key)
            start = end_offset - entry.msg_count + 1
            resolved = ResolvedEntry(
                start_offset=start,
                end_offset=end_offset,
                msg_count=entry.msg_count,
                entry_type=entry.entry_type,
                data_key=entry.data_key,
                encoding=entry.encoding,
                created_at_ms=entry.created_at_ms,
                source="index",
                byte_offset=entry.byte_offset,
                byte_length=entry.byte_length,
            )
            if resolved.start_offset >= stop_before:
                break
            yield resolved

    def _oxia_get(self, key: str, **kwargs: Any) -> Any:
        started = time.perf_counter()
        result = "success"
        try:
            return self._client.get(key, **kwargs)
        except Exception as exc:
            result = self._classify_oxia_result(exc)
            raise
        finally:
            self._observe_oxia_operation(
                operation="get",
                result=result,
                duration_seconds=time.perf_counter() - started,
            )

    def _oxia_range_scan(self, start_key: str, end_key: str, **kwargs: Any) -> Iterator[Any]:
        started = time.perf_counter()
        try:
            iterator = self._client.range_scan(start_key, end_key, **kwargs)
        except Exception as exc:
            self._observe_oxia_operation(
                operation="range_scan",
                result=self._classify_oxia_result(exc),
                duration_seconds=time.perf_counter() - started,
            )
            raise

        def instrumented() -> Iterator[Any]:
            result = "success"
            try:
                for item in iterator:
                    yield item
            except Exception as exc:
                result = self._classify_oxia_result(exc)
                raise
            finally:
                self._observe_oxia_operation(
                    operation="range_scan",
                    result=result,
                    duration_seconds=time.perf_counter() - started,
                )

        return instrumented()

    def _observe_oxia_operation(
        self,
        *,
        operation: str,
        result: str,
        duration_seconds: float,
    ) -> None:
        if self._metrics is None:
            return
        self._metrics.observe_oxia_operation(
            operation=operation,
            result=result,
            duration_seconds=duration_seconds,
        )

    @staticmethod
    def _classify_oxia_result(exc: Exception) -> str:
        if isinstance(exc, oxia.ex.KeyNotFound):
            return "not_found"
        if isinstance(exc, oxia.ex.UnexpectedVersionId):
            return "version_mismatch"
        return "error"

    def _iter_messages_from_entry(
        self,
        tp: TopicPartition,
        entry: ResolvedEntry,
        start_offset: int,
        stop_before: int,
    ) -> Iterator[ReadResult]:
        overlap_start = max(start_offset, entry.start_offset)
        overlap_end = min(stop_before - 1, entry.end_offset)
        if overlap_start > overlap_end:
            return

        blob = self._resolve_entry_blob(entry, self._read_blob(entry))
        for offset in range(overlap_start, overlap_end + 1):
            yield ReadResult(
                topic=tp.topic,
                partition=tp.partition,
                offset=offset,
                payload=self._decode_entry_message(entry, offset, blob),
                entry_type=entry.entry_type,
                start_offset=entry.start_offset,
                end_offset=entry.end_offset,
                data_key=entry.data_key,
                encoding=entry.encoding,
            )

    def _read_blob(self, entry: ResolvedEntry) -> bytes:
        try:
            if entry.byte_offset is not None and entry.byte_length is not None:
                return self._blob_reader.get_range(
                    entry.data_key,
                    entry.byte_offset,
                    entry.byte_length,
                )
            return self._blob_reader.get_bytes(entry.data_key)
        except KeyError as exc:
            raise BlobNotFound(entry.data_key) from exc

    def _decode_entry_message(
        self,
        entry: ResolvedEntry,
        offset: int,
        blob: bytes,
    ) -> bytes:
        if entry.entry_type is EntryType.WAL:
            return self._decode_wal_message(entry, offset, blob)
        if entry.entry_type is EntryType.COMPACTED:
            if self._compacted_reader is not None:
                return self._compacted_reader.read_offset(entry, offset)
            return self._decode_compacted_message(entry, offset, blob)
        raise UnsupportedEntryType(
            f"entry type {entry.entry_type.value} is not readable by this reader"
        )

    @staticmethod
    def _resolve_entry_blob(entry: ResolvedEntry, blob: bytes) -> bytes:
        if entry.byte_offset is None or entry.byte_length is None:
            return blob
        if len(blob) <= entry.byte_length:
            return blob

        end = entry.byte_offset + entry.byte_length
        if entry.byte_offset < 0 or end > len(blob):
            raise MalformedWalBlob(
                "shared WAL byte range falls outside the fetched object body"
            )
        return blob[entry.byte_offset:end]

    def _decode_wal_message(
        self,
        entry: ResolvedEntry,
        offset: int,
        blob: bytes,
    ) -> bytes:
        if entry.encoding == "bytes-batch-v1":
            return self._decode_batch_bytes(entry, offset, blob)
        raise UnsupportedEncoding(f"unsupported WAL encoding {entry.encoding}")

    def _decode_compacted_message(
        self,
        entry: ResolvedEntry,
        offset: int,
        blob: bytes,
    ) -> bytes:
        if entry.encoding == "bytes-batch-v1":
            return self._decode_batch_bytes(entry, offset, blob)
        raise UnsupportedEncoding(f"unsupported compacted encoding {entry.encoding}")

    def _decode_batch_bytes(
        self,
        entry: ResolvedEntry,
        offset: int,
        blob: bytes,
    ) -> bytes:
        record_count, lengths, body = self._decode_batch_layout(blob)
        if record_count != entry.msg_count:
            raise MalformedWalBlob(
                f"batch record count {record_count} does not match metadata {entry.msg_count}"
            )
        record_index = offset - entry.start_offset
        if record_index < 0 or record_index >= record_count:
            raise InconsistentReadState(
                f"offset {offset} is outside batch entry [{entry.start_offset}, {entry.end_offset}]"
            )
        start = sum(lengths[:record_index])
        end = start + lengths[record_index]
        return body[start:end].tobytes()

    @staticmethod
    def _decode_batch_layout(blob: bytes) -> tuple[int, list[int], memoryview]:
        try:
            record_count, lengths, body = decode_batch_layout(blob)
        except ValueError as exc:
            messages = {
                "batch payload is too small": "batch WAL blob is shorter than header",
                "invalid batch payload magic": "batch WAL blob has invalid magic",
                "batch payload is truncated": "batch WAL blob is truncated before lengths table",
            }
            raise MalformedWalBlob(messages.get(str(exc), str(exc))) from exc
        if sum(lengths) != len(body):
            raise MalformedWalBlob("batch WAL blob body length does not match lengths table")
        return record_count, lengths, body

    @staticmethod
    def parse_s3_uri(uri: str) -> tuple[str, str]:
        return parse_s3_uri(uri)

    @staticmethod
    def _json_loads(raw: bytes) -> dict[str, Any]:
        return json_loads(raw)

    @classmethod
    def _json_loads_control(cls, raw: bytes) -> ControlRecord:
        return ControlRecord.from_dict(
            cls._json_loads(raw),
            error_type=InconsistentReadState,
        )

    @staticmethod
    def _optional_env(*names: str) -> str | None:
        for name in names:
            value = os.environ.get(name)
            if value not in (None, ""):
                return value
        return None

    @classmethod
    def _require_env(cls, *names: str) -> str:
        value = cls._optional_env(*names)
        if value is None:
            raise ValueError(f"missing required environment variable: {names[0]}")
        return value

    @staticmethod
    def _format_offset(offset: int) -> str:
        return f"{offset:020d}"

    def _index_lookup_key(self, tp: TopicPartition, offset: int) -> str:
        return self._paths(tp).index_key(offset)

    def _index_scan_end_key(self, tp: TopicPartition) -> str:
        return f"{self._paths(tp).index_prefix}/~"

    @staticmethod
    def _parse_offset_from_index_key(key: str) -> int:
        return int(key.rsplit("/", 1)[1])

    def _tp(self, topic: str, partition: int) -> TopicPartition:
        if not isinstance(topic, str) or topic == "":
            raise ValueError("topic must be a non-empty string")
        if isinstance(partition, bool) or not isinstance(partition, int) or partition < 0:
            raise ValueError("partition must be a non-negative integer")
        return TopicPartition(topic=topic, partition=partition)

    def _paths(self, tp: TopicPartition) -> TopicPartitionPaths:
        return TopicPartitionPaths.build(self._root_prefix, tp)

    @staticmethod
    def _oxia_partition_key(tp: TopicPartition) -> str:
        return f"{tp.topic}:{tp.partition}"


__all__ = [
    "BlobNotFound",
    "BlobReader",
    "CompactedEntryReader",
    "InconsistentReadState",
    "LeaderlessLogReader",
    "LeaderlessLogReaderError",
    "MalformedWalBlob",
    "OffsetOutOfRange",
    "PartitionNotInitialized",
    "ReadResult",
    "ResolvedEntry",
    "S3ClientConfig",
    "S3BlobReader",
    "UnsupportedEncoding",
    "UnsupportedEntryType",
]