# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-serialize-time-series-data-when-the-underlying-storage-uses-variable-length-compression.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-serialize-time-series-data-when-the-underlying-storage-uses-variable-length-compression.warrant.json # Markdown SHA-256: 085ece5e76b6503fa43721ea63f40060d6dfa2ba5b5d363cba70b3a05956cfab # Code AST SHA-256: ea1c811c7e9938df504fad0e8f7bcf7f74bb43a5e5e23d8147921b41c67e1085 # Verified Classes: CompressionAccessVerdict, BlockBoundary, CompressionAccessState, QueryTarget, CompressionAccessManifest, CompressionBlockGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class CompressionAccessVerdict(Enum): RANDOM_ACCESS_SAFE = auto() FULL_BLOCK_SCAN_REQUIRED = auto() OUT_OF_ORDER_WRITE_UNSAFE = auto() @dataclass(frozen=True) class BlockBoundary: block_id: int byte_offset: int timestamp_start: float timestamp_end: float @dataclass(frozen=True) class CompressionAccessState: blocks: tuple[BlockBoundary, ...] has_partial_tail: bool partial_tail_hash: str | None @dataclass(frozen=True) class QueryTarget: target_timestamp: float is_out_of_order_write: bool @dataclass(frozen=True) class CompressionAccessManifest: verdict: CompressionAccessVerdict target_block_id: int | None directive: str manifest_digest: str class CompressionBlockGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_access( cls, state: CompressionAccessState, target: QueryTarget ) -> CompressionAccessManifest: # Gate 1: Out-of-order write check if target.is_out_of_order_write: if not state.has_partial_tail or not state.partial_tail_hash: return CompressionAccessManifest( verdict=CompressionAccessVerdict.OUT_OF_ORDER_WRITE_UNSAFE, target_block_id=None, directive="HALT: Out-of-order write requires uncompressed tail. None available.", manifest_digest=cls._hash("OOO_UNSAFE") ) # Gate 2: Block lookup target_block = None for block in state.blocks: if block.timestamp_start <= target.target_timestamp <= block.timestamp_end: target_block = block break if target_block is None: return CompressionAccessManifest( verdict=CompressionAccessVerdict.FULL_BLOCK_SCAN_REQUIRED, target_block_id=None, directive="HALT: Target timestamp not in any indexed block. Full scan required.", manifest_digest=cls._hash(f"SCAN:{target.target_timestamp}") ) # Gate 3: Verified access digest = cls._hash(json.dumps({ "block": str(target_block.block_id), "offset": str(target_block.byte_offset), "target": f"{target.target_timestamp:.3f}" }, sort_keys=True)) return CompressionAccessManifest( verdict=CompressionAccessVerdict.RANDOM_ACCESS_SAFE, target_block_id=target_block.block_id, directive=f"PROCEED: Seek to block {target_block.block_id} at byte offset {target_block.byte_offset}.", manifest_digest=digest )