From 178ab6c30570228a3b1f324e758aa073d034ac90 Mon Sep 17 00:00:00 2001 From: Jiri Puc Date: Tue, 1 Sep 2026 02:00:33 +0200 Subject: [PATCH 1/3] feat(media): a converter receives a source handle, never the file's bytes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit D104, amending D65's convert(bytes, mime, hints). Design: media_design.md §2.1. A text file fits in memory and a video does not. Under the bytes-first contract the client, the HTTP process, the object store and the converter each materialize a complete copy of the same file, so two concurrent large ingests can exhaust a worker that would have processed either one alone. That is an availability property, not an optimisation. Adds SourceHandlePort with exactly three reads and deliberately no fourth: stream in bounded chunks, read a half-open range [start, end), or materialize to a temporary local file for decoders like FFmpeg that need a seekable path. There is no operation returning the whole source as bytes -- the absence is the load-bearing part: a contract that offers buffering as the convenient option gets buffering, and the failure surfaces on the first large file in production rather than in review. ObjectStorePort grows open_stream and read_range so the handle has something to sit on; read_bytes stays for the small text objects every shipped route converts. Implemented for both the local filesystem and MinIO adapters -- S3 ranges are inclusive on both ends, so the exclusive end becomes end-1 on the wire and every interval in the system stays half-open (D65). Bounds live in materialization, since that is the operation that can exhaust a host: - the caller declares what it can afford and an oversized source is refused before any byte moves -- one integer comparison, not a filled disk; - written bytes are verified against the recorded content hash. An immutable object cannot legitimately change, so a mismatch is corruption or a wrong key, never retryable. Without the check a truncated read becomes a short document that looks complete -- the silent-loss failure D65's coverage rules exist to prevent; - the recorded size is a claim, not a guarantee, so the streaming write also counts what it writes and refuses a source exceeding both the declaration and the accepted bound; and - the temporary file's lifetime belongs to the handle, so a route failing mid-decode cannot leak the file it asked for. One contract serves both deployment shapes: a self-host directory tree and a cloud object store present the same three operations, so no route branches on where bytes live. Nothing consumes the handle yet -- no converter signature changes in this commit and no route behaviour changes. Extending ObjectStorePort required adding the two reads to three test fakes and raising the port-inventory count to thirteen. Validation: ruff format and check clean; pyright 0 errors; 1303 passed, 685 skipped. One pre-existing failure, test_sdk_pushes_lineage_metadata_to_e0, reproduces on the base commit 7c41fdb8 without these changes -- mimetypes on this machine does not resolve .md, so the SDK guesses application/octet-stream. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01XTv3baR8R5EgEGFePkQMjK --- .github/ci/unit-paths.txt | 3 +- decisions.md | 48 ++++++ plan/designs/media_design.md | 61 ++++++- src/rememberstack/adapters/selfhost/minio.py | 50 +++++- .../adapters/selfhost/object_store.py | 22 +++ .../adapters/selfhost/source_handle.py | 126 ++++++++++++++ src/rememberstack/model/__init__.py | 8 + src/rememberstack/model/source_handle.py | 54 ++++++ src/rememberstack/ports/__init__.py | 2 + src/rememberstack/ports/object_store.py | 18 +- src/rememberstack/ports/source_handle.py | 53 ++++++ src/tests/adapters/test_minio_store.py | 18 +- src/tests/adapters/test_source_handle.py | 157 ++++++++++++++++++ .../spine/test_knowledge_control_plane.py | 12 ++ .../test_port_inventory_and_conformance.py | 32 +++- src/tests/workers/test_ingest_admission.py | 13 ++ 16 files changed, 662 insertions(+), 15 deletions(-) create mode 100644 src/rememberstack/adapters/selfhost/source_handle.py create mode 100644 src/rememberstack/model/source_handle.py create mode 100644 src/rememberstack/ports/source_handle.py create mode 100644 src/tests/adapters/test_source_handle.py diff --git a/.github/ci/unit-paths.txt b/.github/ci/unit-paths.txt index a63108a68..e2c4f96cc 100644 --- a/.github/ci/unit-paths.txt +++ b/.github/ci/unit-paths.txt @@ -2,15 +2,16 @@ src/tests/adapters/test_bounded_postgres_read.py src/tests/adapters/test_codex_writer.py src/tests/adapters/test_control_plane_spend_lease.py src/tests/adapters/test_hashed_bearer_auth.py -src/tests/adapters/test_postgres_p1.py src/tests/adapters/test_minio_store.py src/tests/adapters/test_mistral_ocr.py src/tests/adapters/test_openrouter.py +src/tests/adapters/test_postgres_p1.py src/tests/adapters/test_selfhost_forget.py src/tests/adapters/test_selfhost_git.py src/tests/adapters/test_selfhost_purge.py src/tests/adapters/test_selfhost_stores.py src/tests/adapters/test_sentry.py +src/tests/adapters/test_source_handle.py src/tests/adapters/test_telemetry.py src/tests/benchmarks/test_beam_answer_agent.py src/tests/benchmarks/test_beam_official_score.py diff --git a/decisions.md b/decisions.md index 80d5bdf14..584782e7c 100644 --- a/decisions.md +++ b/decisions.md @@ -4793,3 +4793,51 @@ unread; fixing only the sort order while keeping the flag. rest of D21 (connected-components-to-gather, HAC distance-cut, nDR incremental re-decision, `merge_events`, `merged_into`, `resolution_exclusions`) is unchanged. Does not change D95, D99, or D100. +## D104. A converter receives a source handle, never the file's bytes + +**Decision.** The converter contract takes a **source handle** — a read-only +capability on one immutable, already-hashed source — in place of the whole file +as a `bytes` value (amends D65's `convert(bytes, mime, hints)` and, through it, +D38/D57). The handle offers three reads and deliberately no fourth: **stream** +the source in bounded chunks, **read a half-open byte range** `[start, end)`, +or **materialise** it to a temporary local file. There is no operation that +returns the entire source as bytes. `ObjectStorePort` grows the matching +`open_stream` and `read_range` so the handle has something to sit on, and the +existing `read_bytes` stays for the small text objects every shipped route +converts. + +**Context.** A text file fits in memory and a video does not. Under the +bytes-first contract the client, the HTTP process, the object store, and the +converter each materialise a complete copy of the same file, so two concurrent +large ingests can exhaust a worker that would have processed either one alone. +That is an availability property, not an optimisation. The absence of a +whole-file read is the load-bearing part of the decision: a contract that +offers buffering as the convenient option gets buffering, and the failure +surfaces on the first large file in production rather than in review. + +**Bounds live in materialisation**, because that is the operation that can +exhaust a host. The caller declares what it can afford and an oversized source +is refused before any byte moves, costing one comparison rather than a filled +disk. The written bytes are verified against the recorded content hash — an +immutable object cannot legitimately change, so a mismatch is corruption or a +wrong key and never retryable, and without the check a truncated read becomes a +short document that looks complete, which is the silent-loss failure D65's +coverage rules exist to prevent. The recorded size is a claim rather than a +guarantee, so the write also counts what it actually writes and refuses a +source that exceeds both the declaration and the accepted bound. The temporary +file's lifetime belongs to the handle, so a route that fails mid-decode cannot +leak the file it asked for. + +**Consequences.** One contract serves both deployment shapes: a self-host +directory tree and a cloud object store present the same three operations, so +no route branches on where bytes live. Decoders that require a seekable path +are served by bounded materialisation rather than by reimplementing them. +Design home: `plan/designs/media_design.md` §2.1. Interacts with D38 (router), +D57 (Markdown coordinate system), D61 (port boundary), D65 (media routes). + +**Alternatives.** *Keep bytes and raise the worker's memory* — rejected: it +scales the blast radius with source size and solves neither resumability nor +cancellation. *Hand routes a raw provider object* — rejected: it puts a storage +SDK type in the converter contract, which D61 exists to prevent. *Give the +handle a `read_all()` for convenience* — rejected: it becomes the path of least +resistance and reintroduces exactly the failure this decision removes. diff --git a/plan/designs/media_design.md b/plan/designs/media_design.md index 743571a87..0fdfa8f5c 100644 --- a/plan/designs/media_design.md +++ b/plan/designs/media_design.md @@ -103,7 +103,8 @@ Notes an implementer needs: (the route prompt/adapter enforces it), and it is what makes §5's disclosure a property of *ranges the converter wrote*, not a per-claim judgment anyone has to make downstream. - **The generalized converter contract** (refines D38 a second time): - `convert(bytes, mime, hints) → { document.md, source_map, derived_assets[], manifest }` — + `convert(source, mime, hints) → { document.md, source_map, derived_assets[], manifest }` — + where `source` is a **source handle** (§2.1), not the file's bytes — the *page map* generalizes to a **source map** (§4), `derived_assets` are the `media/` children with their locators, and the `manifest` is the route's **complete self-account**, with required fields (nullable only where a capability is genuinely absent): @@ -118,6 +119,64 @@ Notes an implementer needs: regions the tool could not read) — a conversion that silently drops ten minutes of a recording is the same lie as a silent top-k. + +### 2.1 The source handle — why a converter is not handed bytes + +*Decision: D104.* + +A text file fits in memory, so the original converter contract took the whole +file as a `bytes` value. A video does not. If the contract keeps demanding +bytes, then the client, the HTTP process, the object store, and the converter +each hold a complete copy of the same multi-gigabyte file, and the host runs +out of memory before any model work starts. That is a correctness and +availability property, not an optimisation: two concurrent large ingests can +take down a worker that would happily have processed either one alone. + +So a route receives a **source handle** — a read-only capability on one +immutable, already-hashed source — and decides for itself how much of it to +bring into memory. The handle offers exactly three ways to read, and +deliberately no fourth: + +- **stream it** — receive the source in order, one bounded chunk at a time, + which is what hashing, copying, and demuxing need; +- **read a range** — take the half-open byte interval `[start, end)`, which is + what container parsing needs to read a header or an index without the body; + half-open because every other interval in the system is (§4), so nobody has + to remember which end is inclusive; and +- **materialise it** — write the source to a temporary local file and hand + over the path. Decoders like FFmpeg want a seekable file rather than a + stream, and refusing to provide one would mean reimplementing them. + +There is deliberately **no method that returns the whole source as bytes**. +The absence is the point: a contract that offers whole-file buffering as the +convenient option gets whole-file buffering, and the failure only shows up on +the first large file in production. + +Materialisation is where the bounds live, because it is the operation that can +actually exhaust a host. Three properties are enforced by the handle rather +than trusted to each route: + +- **the caller declares what it can afford**, and a source larger than that is + refused before a single byte moves — an oversized source costs one integer + comparison, not a filled disk; +- **the written bytes are verified against the source's content hash.** An + immutable object cannot legitimately change, so a mismatch is corruption or + a wrong key, never a retryable condition. Without this check a truncated + read becomes a short document that looks complete, which is precisely the + silent-loss failure the coverage rules in §2 exist to prevent; and +- **the file is removed when the route is done with it**, whether it returned + or raised. A converter that fails mid-decode cannot leak the file it asked + for, because it never owned the file's lifetime. + +The recorded size is a claim, not a guarantee, so the streaming write also +counts what it actually writes and refuses if the source turns out to be +larger than both the declaration and the accepted bound. + +One handle contract serves both deployment shapes. A self-host deployment +reads from a directory tree and a cloud deployment reads from object storage, +but a route sees the same three operations either way, so no route contains a +branch on where the bytes live. + ## 3. What "already works" and stays untouched The representation flows the standard pipeline with **no media-specific machinery**: blocks diff --git a/src/rememberstack/adapters/selfhost/minio.py b/src/rememberstack/adapters/selfhost/minio.py index c529bf68e..fba2c5eec 100644 --- a/src/rememberstack/adapters/selfhost/minio.py +++ b/src/rememberstack/adapters/selfhost/minio.py @@ -1,5 +1,6 @@ """S3-compatible MinIO object storage for the self-host profile.""" +from collections.abc import Iterator from typing import cast from typing import NotRequired from typing import Protocol @@ -15,6 +16,7 @@ from rememberstack.model import ObjectAlreadyExistsError from rememberstack.model import ObjectKey from rememberstack.model import ObjectKeyEscapesRootError +from rememberstack.model import SourceRangeError class MinIOSettings(BaseSettings): @@ -31,8 +33,8 @@ class MinIOSettings(BaseSettings): class _StreamingBody(Protocol): """The two response-body operations used by this adapter.""" - def read(self) -> bytes: - """Read the complete response body.""" + def read(self, amt: int | None = None) -> bytes: + """Read the whole body, or at most ``amt`` bytes when given.""" ... def close(self) -> None: @@ -77,8 +79,10 @@ def create_bucket(self, *, Bucket: str) -> object: """Create one bucket.""" ... - def get_object(self, *, Bucket: str, Key: str) -> _GetObjectOutput: - """Read one object.""" + def get_object( + self, *, Bucket: str, Key: str, Range: str | None = None + ) -> _GetObjectOutput: + """Read one object, optionally restricted to an HTTP byte range.""" ... def put_object( @@ -154,6 +158,44 @@ def read_bytes(self, *, key: ObjectKey) -> bytes: finally: body.close() + def open_stream( + self, *, key: ObjectKey, chunk_bytes: int = 1024 * 1024 + ) -> Iterator[bytes]: + """Yield one object in order, holding at most one chunk at a time.""" + if chunk_bytes <= 0: + raise SourceRangeError(f"chunk_bytes must be positive, got {chunk_bytes}") + response = self._client.get_object( + Bucket=self._bucket, Key=_validated_key(key=key) + ) + body = response["Body"] + try: + while chunk := body.read(chunk_bytes): + yield chunk + finally: + body.close() + + def read_range(self, *, key: ObjectKey, start: int, end: int) -> bytes: + """Read the half-open byte interval ``[start, end)`` of one object. + + S3 ranges are inclusive on both ends, so the exclusive `end` becomes + ``end - 1`` on the wire. Converting here keeps every interval in the + system half-open (D65) regardless of the provider behind the port. + """ + if start < 0 or end <= start: + raise SourceRangeError( + f"range [{start}, {end}) is empty, reversed, or negative" + ) + response = self._client.get_object( + Bucket=self._bucket, + Key=_validated_key(key=key), + Range=f"bytes={start}-{end - 1}", + ) + body = response["Body"] + try: + return body.read() + finally: + body.close() + def write_bytes( self, *, key: ObjectKey, content: bytes, storage_class: str | None = None ) -> None: diff --git a/src/rememberstack/adapters/selfhost/object_store.py b/src/rememberstack/adapters/selfhost/object_store.py index 18833beb1..7c4a06a85 100644 --- a/src/rememberstack/adapters/selfhost/object_store.py +++ b/src/rememberstack/adapters/selfhost/object_store.py @@ -1,10 +1,12 @@ """Local-filesystem object store adapter: immutable bytes under one root (D61/D62).""" +from collections.abc import Iterator from pathlib import Path from rememberstack.model import ObjectAlreadyExistsError from rememberstack.model import ObjectKey from rememberstack.model import ObjectKeyEscapesRootError +from rememberstack.model import SourceRangeError class LocalFSObjectStore: @@ -19,6 +21,26 @@ def read_bytes(self, *, key: ObjectKey) -> bytes: """Read all bytes stored under an existing object key.""" return self._path_for(key=key).read_bytes() + def open_stream( + self, *, key: ObjectKey, chunk_bytes: int = 1024 * 1024 + ) -> Iterator[bytes]: + """Yield one object in order, holding at most one chunk at a time.""" + if chunk_bytes <= 0: + raise SourceRangeError(f"chunk_bytes must be positive, got {chunk_bytes}") + with self._path_for(key=key).open(mode="rb") as handle: + while chunk := handle.read(chunk_bytes): + yield chunk + + def read_range(self, *, key: ObjectKey, start: int, end: int) -> bytes: + """Read the half-open byte interval ``[start, end)`` of one object.""" + if start < 0 or end <= start: + raise SourceRangeError( + f"range [{start}, {end}) is empty, reversed, or negative" + ) + with self._path_for(key=key).open(mode="rb") as handle: + handle.seek(start) + return handle.read(end - start) + def write_bytes( self, *, key: ObjectKey, content: bytes, storage_class: str | None = None ) -> None: diff --git a/src/rememberstack/adapters/selfhost/source_handle.py b/src/rememberstack/adapters/selfhost/source_handle.py new file mode 100644 index 000000000..a78bba759 --- /dev/null +++ b/src/rememberstack/adapters/selfhost/source_handle.py @@ -0,0 +1,126 @@ +"""A bounded source handle over any object store (D61 seam, media-safe reads). + +`ObjectSourceHandle` is the adapter that lets a converter read a large source +without the whole file existing in the worker's heap. It works over any +`ObjectStorePort`, so the same converter code runs against a local directory +in a self-host deployment and against S3 in a managed one. +""" + +from collections.abc import Iterator +from contextlib import contextmanager +import hashlib +from pathlib import Path +import tempfile + +from rememberstack.model import SourceHashMismatchError +from rememberstack.model import SourceIdentity +from rememberstack.model import SourceRangeError +from rememberstack.model import SourceTooLargeError +from rememberstack.ports import ObjectStorePort + +_HASH_CHUNK_BYTES = 1024 * 1024 + + +class ObjectSourceHandle: + """Bounded reads of one immutable source held in an object store.""" + + def __init__( + self, + *, + store: ObjectStorePort, + identity: SourceIdentity, + temp_root: Path | None = None, + ) -> None: + """Bind the handle to one source and the store that holds its bytes. + + `temp_root` is where `materialize_seekable` writes. Leaving it None + uses the platform temporary directory; a deployment that bounds + worker disk points it at the volume it actually sized. + """ + self._store = store + self._identity = identity + self._temp_root = temp_root + + @property + def identity(self) -> SourceIdentity: + """The source's recorded key, content hash, size, and declared type.""" + return self._identity + + def open_stream(self, *, chunk_bytes: int = _HASH_CHUNK_BYTES) -> Iterator[bytes]: + """Yield the source in order, holding at most one chunk at a time.""" + return self._store.open_stream( + key=self._identity.object_key, chunk_bytes=chunk_bytes + ) + + def read_range(self, *, start: int, end: int) -> bytes: + """Read the half-open byte interval ``[start, end)`` of the source. + + The range is checked against the size recorded at ingest, so a caller + asking past the end is refused here rather than receiving a short read + it might mistake for the end of meaningful content. + """ + if start < 0 or end <= start: + raise SourceRangeError( + f"range [{start}, {end}) is empty, reversed, or negative" + ) + if end > self._identity.byte_size: + raise SourceRangeError( + f"range [{start}, {end}) extends past the recorded source size " + f"{self._identity.byte_size}" + ) + return self._store.read_range( + key=self._identity.object_key, start=start, end=end + ) + + @contextmanager + def materialize_seekable(self, *, max_bytes: int) -> Iterator[Path]: + """Stream the source to a temporary file, removed when the block exits. + + Three properties matter and all three are enforced here rather than + trusted to the caller: + + * the bound is checked against the recorded size *before* any bytes + move, so an oversized source costs one comparison, not a filled disk; + * the written bytes are hashed and compared to the source's content + hash, so a truncated or corrupted read fails loudly instead of being + converted into a plausible-looking short document; and + * the file is removed on the way out whether the body raised or not. + """ + if max_bytes <= 0: + raise SourceTooLargeError(f"max_bytes must be positive, got {max_bytes}") + if self._identity.byte_size > max_bytes: + raise SourceTooLargeError( + f"source {self._identity.object_key.root!r} is " + f"{self._identity.byte_size} bytes, over the {max_bytes} accepted" + ) + if self._temp_root is not None: + self._temp_root.mkdir(parents=True, exist_ok=True) + directory = tempfile.mkdtemp( + prefix="rememberstack-source-", + dir=None if self._temp_root is None else str(self._temp_root), + ) + path = Path(directory) / "source" + try: + digest = hashlib.sha256() + written = 0 + with path.open(mode="wb") as handle: + for chunk in self.open_stream(): + written += len(chunk) + if written > max_bytes: + raise SourceTooLargeError( + f"source {self._identity.object_key.root!r} exceeded the " + f"{max_bytes} accepted while streaming; the recorded size " + f"{self._identity.byte_size} was wrong" + ) + digest.update(chunk) + handle.write(chunk) + if digest.hexdigest() != self._identity.content_hash: + raise SourceHashMismatchError( + f"source {self._identity.object_key.root!r} hashed to " + f"{digest.hexdigest()}, not the recorded " + f"{self._identity.content_hash}" + ) + yield path + finally: + path.unlink(missing_ok=True) + Path(directory).rmdir() diff --git a/src/rememberstack/model/__init__.py b/src/rememberstack/model/__init__.py index 073072ba7..f243b2bf5 100644 --- a/src/rememberstack/model/__init__.py +++ b/src/rememberstack/model/__init__.py @@ -346,6 +346,10 @@ from rememberstack.model.sections import SkeletonVerdict from rememberstack.model.sections import SnappedSection from rememberstack.model.sections import StructureRouteTag +from rememberstack.model.source_handle import SourceHashMismatchError +from rememberstack.model.source_handle import SourceIdentity +from rememberstack.model.source_handle import SourceRangeError +from rememberstack.model.source_handle import SourceTooLargeError from rememberstack.model.spend_lease import SpendLeaseRefused from rememberstack.model.spend_lease import SpendLeaseUnavailable from rememberstack.model.telemetry import TelemetryAttribute @@ -499,6 +503,10 @@ "ObjectAlreadyExistsError", "ObjectKey", "ObjectKeyEscapesRootError", + "SourceHashMismatchError", + "SourceIdentity", + "SourceRangeError", + "SourceTooLargeError", "ObservationAssertion", "ObservationCandidate", "ObservationForEmbedding", diff --git a/src/rememberstack/model/source_handle.py b/src/rememberstack/model/source_handle.py new file mode 100644 index 000000000..934baeb93 --- /dev/null +++ b/src/rememberstack/model/source_handle.py @@ -0,0 +1,54 @@ +"""Identity and failure types for bounded reads of one immutable source. + +A media source is too large to carry as `bytes`. These types describe what a +converter is allowed to know about a source it has not read, and how a read +refuses when it would exceed the bounds the caller accepted. +""" + +from typing import Annotated + +from pydantic import BaseModel +from pydantic import ConfigDict +from pydantic import Field + +from rememberstack.model.object_store import ObjectKey + +Sha256Hex = Annotated[str, Field(pattern=r"^[0-9a-f]{64}$")] + + +class SourceIdentity(BaseModel): + """What a converter may know about a source without reading it. + + The hash and size are the E0 facts recorded when the raw object was + written, not values a converter or a caller supplies. `mime` is the + caller's declared type and stays an untrusted hint: routing and rating + belong to a structural probe of the actual bytes, never to this field. + """ + + model_config = ConfigDict(frozen=True, extra="forbid") + + object_key: ObjectKey + content_hash: Sha256Hex + byte_size: int = Field(ge=0) + mime: str = Field(min_length=1) + + +class SourceTooLargeError(Exception): + """A read or materialization would exceed the bound the caller accepted. + + Raised *before* the bytes are moved. A converter declares what it can + afford and is refused up front, rather than discovering the cost by + exhausting the worker. + """ + + +class SourceHashMismatchError(Exception): + """Materialized bytes did not hash to the source's recorded content hash. + + Immutable objects cannot legitimately change, so this is corruption or a + wrong key — never a retryable condition on the same inputs. + """ + + +class SourceRangeError(Exception): + """A byte range that is empty, reversed, or outside the source.""" diff --git a/src/rememberstack/ports/__init__.py b/src/rememberstack/ports/__init__.py index f0863b551..195075416 100644 --- a/src/rememberstack/ports/__init__.py +++ b/src/rememberstack/ports/__init__.py @@ -11,6 +11,7 @@ from rememberstack.ports.purge import ObjectPurgePort from rememberstack.ports.purge import ProjectionPurgePort from rememberstack.ports.queue import TaskQueuePort +from rememberstack.ports.source_handle import SourceHandlePort from rememberstack.ports.telemetry import TelemetryPort __all__ = ( @@ -24,6 +25,7 @@ "PostgresReadPoolPort", "ObjectPurgePort", "ProjectionPurgePort", + "SourceHandlePort", "TaskQueuePort", "TelemetryPort", ) diff --git a/src/rememberstack/ports/object_store.py b/src/rememberstack/ports/object_store.py index 918037b41..d3676fd55 100644 --- a/src/rememberstack/ports/object_store.py +++ b/src/rememberstack/ports/object_store.py @@ -1,5 +1,6 @@ """D61 byte/object-key seam for immutable raw inputs, artifacts, and snapshots.""" +from collections.abc import Iterator from typing import Protocol from typing import runtime_checkable @@ -11,7 +12,22 @@ class ObjectStorePort(Protocol): """Read and create immutable objects without exposing storage-provider types.""" def read_bytes(self, *, key: ObjectKey) -> bytes: - """Read all bytes stored under an existing object key.""" + """Read all bytes stored under an existing object key. + + Correct for the small text objects every shipped route converts. + A caller that cannot bound the object's size uses `open_stream` or + `read_range` instead — see `SourceHandlePort`. + """ + ... + + def open_stream( + self, *, key: ObjectKey, chunk_bytes: int = 1024 * 1024 + ) -> Iterator[bytes]: + """Yield one object in order, holding at most one chunk at a time.""" + ... + + def read_range(self, *, key: ObjectKey, start: int, end: int) -> bytes: + """Read the half-open byte interval ``[start, end)`` of one object.""" ... def write_bytes( diff --git a/src/rememberstack/ports/source_handle.py b/src/rememberstack/ports/source_handle.py new file mode 100644 index 000000000..97752c3e2 --- /dev/null +++ b/src/rememberstack/ports/source_handle.py @@ -0,0 +1,53 @@ +"""Bounded read access to one immutable source, without buffering it whole. + +The converter contract begins with complete `bytes` (D38/D57), which is +workable for text and unworkable for media: the client, the HTTP process, the +object store, and the converter each materialize the whole file. This port is +the seam that lets a route read a source by stream, by range, or as a bounded +local file, so a large source costs bounded memory rather than its own size. +""" + +from collections.abc import Iterator +from contextlib import AbstractContextManager +from pathlib import Path +from typing import Protocol +from typing import runtime_checkable + +from rememberstack.model import SourceIdentity + + +@runtime_checkable +class SourceHandlePort(Protocol): + """Read one immutable source without requiring its size in memory. + + Deliberately absent: any method returning the whole source as `bytes`. + A caller that genuinely needs every byte streams them and decides what to + keep; the port never makes whole-file buffering the easy path. + """ + + @property + def identity(self) -> SourceIdentity: + """The source's recorded key, content hash, size, and declared type.""" + ... + + def open_stream(self, *, chunk_bytes: int = 1024 * 1024) -> Iterator[bytes]: + """Yield the source in order, holding at most one chunk at a time.""" + ... + + def read_range(self, *, start: int, end: int) -> bytes: + """Read the half-open byte interval ``[start, end)``. + + Half-open to match every other interval in the system (D65 locators), + so a caller never has to remember which end is inclusive. + """ + ... + + def materialize_seekable(self, *, max_bytes: int) -> AbstractContextManager[Path]: + """Write the source to a temporary local file, removed on exit. + + Decoders such as FFmpeg need a seekable path, not a stream. The bound + is checked against the recorded size before any bytes move, and the + context manager owns cleanup so a failing converter cannot leak the + file it asked for. + """ + ... diff --git a/src/tests/adapters/test_minio_store.py b/src/tests/adapters/test_minio_store.py index 8156814e8..3ddf55396 100644 --- a/src/tests/adapters/test_minio_store.py +++ b/src/tests/adapters/test_minio_store.py @@ -22,9 +22,9 @@ def __init__(self, *, content: bytes) -> None: self._stream = BytesIO(content) self.closed = False - def read(self) -> bytes: - """Read all remaining bytes.""" - return self._stream.read() + def read(self, amt: int | None = None) -> bytes: + """Read all remaining bytes, or at most ``amt`` when given.""" + return self._stream.read() if amt is None else self._stream.read(amt) def close(self) -> None: """Record connection release.""" @@ -52,9 +52,15 @@ def create_bucket(self, *, Bucket: str) -> object: self.buckets.add(Bucket) return {} - def get_object(self, *, Bucket: str, Key: str) -> _GetObjectOutput: - """Return one streaming body.""" - body = _Body(content=self.objects[(Bucket, Key)][0]) + def get_object( + self, *, Bucket: str, Key: str, Range: str | None = None + ) -> _GetObjectOutput: + """Return one streaming body, honoring an inclusive HTTP byte range.""" + content = self.objects[(Bucket, Key)][0] + if Range is not None: + first, _, last = Range.removeprefix("bytes=").partition("-") + content = content[int(first) : int(last) + 1] + body = _Body(content=content) self.last_body = body return {"Body": body} diff --git a/src/tests/adapters/test_source_handle.py b/src/tests/adapters/test_source_handle.py new file mode 100644 index 000000000..b60e4309c --- /dev/null +++ b/src/tests/adapters/test_source_handle.py @@ -0,0 +1,157 @@ +"""Bounded source reads: streaming, ranges, size refusal, and hash verification.""" + +import hashlib +from pathlib import Path + +import pytest + +from rememberstack.adapters.selfhost.object_store import LocalFSObjectStore +from rememberstack.adapters.selfhost.source_handle import ObjectSourceHandle +from rememberstack.model import ObjectKey +from rememberstack.model import SourceHashMismatchError +from rememberstack.model import SourceIdentity +from rememberstack.model import SourceRangeError +from rememberstack.model import SourceTooLargeError + +_KEY = ObjectKey("raw/recording.mp4") + + +def _handle( + *, tmp_path: Path, content: bytes, declared_size: int | None = None +) -> ObjectSourceHandle: + """Build a handle over a local store holding `content` at the fixed key.""" + store = LocalFSObjectStore(root=tmp_path / "objects") + store.write_bytes(key=_KEY, content=content) + return ObjectSourceHandle( + store=store, + identity=SourceIdentity( + object_key=_KEY, + content_hash=hashlib.sha256(content).hexdigest(), + byte_size=len(content) if declared_size is None else declared_size, + mime="video/mp4", + ), + temp_root=tmp_path / "work", + ) + + +def test_stream_reassembles_the_source_without_whole_file_chunks( + tmp_path: Path, +) -> None: + """Streaming yields the whole source in bounded pieces, in order.""" + content = bytes(range(256)) * 40 + handle = _handle(tmp_path=tmp_path, content=content) + + chunks = list(handle.open_stream(chunk_bytes=1024)) + + assert b"".join(chunks) == content + assert max(len(chunk) for chunk in chunks) <= 1024 + assert len(chunks) > 1, "a bounded read of a large source must be split" + + +def test_range_reads_are_half_open(tmp_path: Path) -> None: + """``[start, end)`` matches every other interval in the system (D65).""" + handle = _handle(tmp_path=tmp_path, content=b"0123456789") + + assert handle.read_range(start=2, end=5) == b"234" + + +@pytest.mark.parametrize( + ("start", "end"), [(5, 5), (5, 2), (-1, 4)], ids=["empty", "reversed", "negative"] +) +def test_degenerate_ranges_are_refused(tmp_path: Path, start: int, end: int) -> None: + """An empty, reversed, or negative range is a caller bug, not a short read.""" + handle = _handle(tmp_path=tmp_path, content=b"0123456789") + + with pytest.raises(SourceRangeError): + handle.read_range(start=start, end=end) + + +def test_range_past_the_recorded_size_is_refused(tmp_path: Path) -> None: + """Reading past the end must fail loudly, not return a silent short read. + + A converter that mistakes a short read for the end of content produces a + truncated document that looks complete — the failure D65 coverage rules + exist to prevent. + """ + handle = _handle(tmp_path=tmp_path, content=b"0123456789") + + with pytest.raises(SourceRangeError): + handle.read_range(start=8, end=99) + + +def test_materialize_gives_a_seekable_path_and_removes_it(tmp_path: Path) -> None: + """Decoders need a real file; the handle owns its lifetime, not the caller.""" + content = b"video-bytes" * 100 + handle = _handle(tmp_path=tmp_path, content=content) + + with handle.materialize_seekable(max_bytes=len(content)) as path: + assert path.read_bytes() == content + materialized = path + + assert not materialized.exists() + assert not materialized.parent.exists() + + +def test_materialize_removes_the_file_even_when_the_caller_raises( + tmp_path: Path, +) -> None: + """A converter that fails mid-decode must not leak the file it asked for.""" + handle = _handle(tmp_path=tmp_path, content=b"payload") + materialized: Path | None = None + + with pytest.raises(RuntimeError): + with handle.materialize_seekable(max_bytes=1024) as path: + materialized = path + raise RuntimeError("decoder exploded") + + assert materialized is not None, "the body must have run before it raised" + assert not materialized.exists() + + +def test_oversized_source_is_refused_before_any_bytes_move(tmp_path: Path) -> None: + """The bound is checked against the recorded size, so refusal costs nothing.""" + handle = _handle(tmp_path=tmp_path, content=b"x" * 5000) + + with pytest.raises(SourceTooLargeError): + with handle.materialize_seekable(max_bytes=100): + pytest.fail("an oversized source must never be materialized") + + assert not (tmp_path / "work").exists() or not any((tmp_path / "work").iterdir()) + + +def test_understated_size_is_still_caught_while_streaming(tmp_path: Path) -> None: + """A wrong recorded size cannot be used to smuggle past the accepted bound.""" + content = b"y" * 5000 + handle = _handle(tmp_path=tmp_path, content=content, declared_size=10) + + with pytest.raises(SourceTooLargeError): + with handle.materialize_seekable(max_bytes=100): + pytest.fail("the streaming guard must fire when the size was wrong") + + +def test_corrupted_bytes_fail_the_hash_check(tmp_path: Path) -> None: + """Immutable objects cannot legitimately change, so a mismatch is terminal.""" + store = LocalFSObjectStore(root=tmp_path / "objects") + store.write_bytes(key=_KEY, content=b"actual-content") + handle = ObjectSourceHandle( + store=store, + identity=SourceIdentity( + object_key=_KEY, + content_hash=hashlib.sha256(b"different-content").hexdigest(), + byte_size=len(b"actual-content"), + mime="video/mp4", + ), + temp_root=tmp_path / "work", + ) + + with pytest.raises(SourceHashMismatchError): + with handle.materialize_seekable(max_bytes=1024): + pytest.fail("a hash mismatch must not reach the converter") + + +def test_nonpositive_chunk_size_is_refused(tmp_path: Path) -> None: + """A zero chunk would spin forever; a negative one is meaningless.""" + handle = _handle(tmp_path=tmp_path, content=b"data") + + with pytest.raises(SourceRangeError): + list(handle.open_stream(chunk_bytes=0)) diff --git a/src/tests/spine/test_knowledge_control_plane.py b/src/tests/spine/test_knowledge_control_plane.py index ab5129b86..d097ec2c4 100644 --- a/src/tests/spine/test_knowledge_control_plane.py +++ b/src/tests/spine/test_knowledge_control_plane.py @@ -552,6 +552,18 @@ def read_bytes(self, *, key: ObjectKey) -> bytes: """Return one previously archived session transcript.""" return self.objects[key.root] + def open_stream( + self, *, key: ObjectKey, chunk_bytes: int = 1024 * 1024 + ) -> Iterator[bytes]: + """Yield the stored bytes in fixed-size chunks.""" + content = self.read_bytes(key=key) + for offset in range(0, len(content), chunk_bytes): + yield content[offset : offset + chunk_bytes] + + def read_range(self, *, key: ObjectKey, start: int, end: int) -> bytes: + """Return the half-open byte interval of the stored bytes.""" + return self.read_bytes(key=key)[start:end] + def write_bytes( self, *, key: ObjectKey, content: bytes, storage_class: str | None = None ) -> None: diff --git a/src/tests/test_port_inventory_and_conformance.py b/src/tests/test_port_inventory_and_conformance.py index a545ba139..75546f95e 100644 --- a/src/tests/test_port_inventory_and_conformance.py +++ b/src/tests/test_port_inventory_and_conformance.py @@ -1,5 +1,6 @@ """Inventory tests for D61 substrate seams plus D74 store capabilities.""" +from collections.abc import Iterator from datetime import datetime from datetime import timezone from decimal import Decimal @@ -12,6 +13,7 @@ from pydantic import BaseModel from pydantic import SecretBytes +from rememberstack.adapters.selfhost.source_handle import ObjectSourceHandle from rememberstack.model import AuthenticatedContext from rememberstack.model import EmbeddingRequest from rememberstack.model import EmbeddingResponse @@ -23,6 +25,7 @@ from rememberstack.model import ProviderCallUsage from rememberstack.model import PublishedMounts from rememberstack.model import QueueRoute +from rememberstack.model import SourceIdentity from rememberstack.model import StructuredResponseModel from rememberstack.model import TelemetryEvent from rememberstack.model import UTCDateTime @@ -32,6 +35,7 @@ from rememberstack.ports import ModelProviderPort from rememberstack.ports import MountPublisherPort from rememberstack.ports import ObjectStorePort +from rememberstack.ports import SourceHandlePort from rememberstack.ports import TaskQueuePort from rememberstack.ports import TelemetryPort import rememberstack.ports.auth as auth_module @@ -43,6 +47,7 @@ import rememberstack.ports.postgres_read as postgres_read_module import rememberstack.ports.purge as purge_module import rememberstack.ports.queue as queue_module +import rememberstack.ports.source_handle as source_handle_module import rememberstack.ports.telemetry as telemetry_module ResponseT = TypeVar("ResponseT", bound=StructuredResponseModel) @@ -57,6 +62,7 @@ postgres_read_module, queue_module, purge_module, + source_handle_module, telemetry_module, ) _PORT_EXPORTS = { @@ -70,6 +76,7 @@ "ObjectPurgePort", "PostgresReadPoolPort", "ProjectionPurgePort", + "SourceHandlePort", "TaskQueuePort", "TelemetryPort", } @@ -92,6 +99,18 @@ def read_bytes(self, *, key: ObjectKey) -> bytes: """Return the bytes stored under the requested key.""" return self.objects[key.root] + def open_stream( + self, *, key: ObjectKey, chunk_bytes: int = 1024 * 1024 + ) -> Iterator[bytes]: + """Yield the stored bytes in fixed-size chunks.""" + content = self.objects[key.root] + for offset in range(0, len(content), chunk_bytes): + yield content[offset : offset + chunk_bytes] + + def read_range(self, *, key: ObjectKey, start: int, end: int) -> bytes: + """Return the half-open byte interval of the stored bytes.""" + return self.objects[key.root][start:end] + def write_bytes( self, *, key: ObjectKey, content: bytes, storage_class: str | None = None ) -> None: @@ -206,6 +225,15 @@ def announce( _object_store_assignment: ObjectStorePort = FakeObjectStore() +_source_handle_assignment: SourceHandlePort = ObjectSourceHandle( + store=FakeObjectStore(), + identity=SourceIdentity( + object_key=ObjectKey("raw/example"), + content_hash="0" * 64, + byte_size=0, + mime="video/mp4", + ), +) _mount_assignment: MountPublisherPort = FakeMountPublisher() _git_assignment: KGitRemotePort = FakeKGitRemote() _model_assignment: ModelProviderPort = FakeModelProvider() @@ -228,11 +256,11 @@ def _defined_protocols() -> set[type[object]]: return result -def test_inventory_exports_exactly_twelve_defined_protocols() -> None: +def test_inventory_exports_exactly_thirteen_defined_protocols() -> None: """Keep D61/D74 seams plus bounded PostgreSQL read admission explicit.""" assert set(ports.__all__) == _PORT_EXPORTS assert {protocol.__name__ for protocol in _defined_protocols()} == _PORT_EXPORTS - assert len(_defined_protocols()) == 12 + assert len(_defined_protocols()) == 13 def test_representative_fakes_conform_structurally() -> None: diff --git a/src/tests/workers/test_ingest_admission.py b/src/tests/workers/test_ingest_admission.py index ac00ae989..d75ccd739 100644 --- a/src/tests/workers/test_ingest_admission.py +++ b/src/tests/workers/test_ingest_admission.py @@ -1,5 +1,6 @@ """Fast D74 proof that ingest checks admission before persisting bytes.""" +from collections.abc import Iterator from datetime import datetime from datetime import timedelta from datetime import timezone @@ -56,6 +57,18 @@ def __init__(self) -> None: def read_bytes(self, *, key: ObjectKey) -> bytes: raise AssertionError(f"unexpected read of {key.root}") + def open_stream( + self, *, key: ObjectKey, chunk_bytes: int = 1024 * 1024 + ) -> Iterator[bytes]: + """Yield the stored bytes in fixed-size chunks.""" + content = self.read_bytes(key=key) + for offset in range(0, len(content), chunk_bytes): + yield content[offset : offset + chunk_bytes] + + def read_range(self, *, key: ObjectKey, start: int, end: int) -> bytes: + """Return the half-open byte interval of the stored bytes.""" + return self.read_bytes(key=key)[start:end] + def write_bytes( self, *, key: ObjectKey, content: bytes, storage_class: str | None = None ) -> None: From a69e332d00378391b9e0a9eefe45d00fe0044ac8 Mon Sep 17 00:00:00 2001 From: Jiri Puc Date: Tue, 1 Sep 2026 02:10:33 +0200 Subject: [PATCH 2/3] =?UTF-8?q?fix(media):=20address=20review=20=E2=80=94?= =?UTF-8?q?=20cleanup,=20short=20reads,=20and=20a=20bounded=20whole=20read?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit External review returned BLOCK with eight findings. All addressed. BLOCK — teardown used unlink plus rmdir, which raises "directory not empty" the moment a decoder writes anything beside the source it was handed: an index, a sidecar, a split audio track. That both leaked the directory and masked whatever the route was already raising. Now shutil.rmtree with ignore_errors, so cleanup cannot become the failure. BLOCK — reads could come back short. LocalFS returned a truncated slice past EOF, MinIO let botocore's ClientError escape on a 416 (a provider type in a port D61 keeps provider-free) and returned a short 206 body otherwise, and the handle validated only against the recorded size, never against what actually arrived. A stored object shorter than its record therefore produced a silent short read. All three now return exactly the requested bytes or raise SourceRangeError, and 416 is translated at the adapter. This is the failure the range contract exists to prevent: a container parser handed a truncated header cannot distinguish it from a valid one and produces confident nonsense instead of an error. MAJOR — max_bytes had to be positive, so a zero-byte source could not be materialized under a zero bound; a negative bound now raises ValueError as the caller bug it is, rather than being reported as a size refusal. MAJOR — MinIO's open_stream and read_range had no tests at all. The fakes were updated but nothing exercised the wire translation, so an off-by-one in the inclusive-range conversion would have passed silently. Five tests added, including one asserting [2, 5) reaches S3 as bytes=2-4. MINOR — the materialized file is named with the source's own suffix. Demuxers sniff the container from the extension before falling back to probing, so a bare "source" costs a class of cannot-open bug for nothing. MINOR — added read_bounded(max_bytes=...). The review's point stands: with no in-memory read at all, a text route writes b"".join(open_stream()), which is an unbounded read wearing a bounded read's clothes. The design intent was never "no bytes in memory" but "no bytes in memory without someone deciding how many", so the fourth read requires the caller to name a limit. Design and D104 corrected — they claimed three reads and no fourth, which is no longer true. MINOR — tests added for zero-byte sources, temp_root=None, cleanup with a sidecar present, a truncated stored object, and two concurrent live materializations not sharing or co-deleting a directory. Validation: ruff clean; pyright 0 errors; test inventory OK; 1319 passed, 687 skipped. The one failure, test_sdk_pushes_lineage_metadata_to_e0, reproduces on the base commit without these changes — mimetypes on this machine does not resolve .md. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01XTv3baR8R5EgEGFePkQMjK --- decisions.md | 26 ++-- plan/designs/media_design.md | 29 +++-- src/rememberstack/adapters/selfhost/minio.py | 30 ++++- .../adapters/selfhost/object_store.py | 8 +- .../adapters/selfhost/source_handle.py | 44 ++++++- src/rememberstack/ports/source_handle.py | 14 +++ src/tests/adapters/test_minio_store.py | 62 ++++++++++ src/tests/adapters/test_source_handle.py | 116 ++++++++++++++++++ 8 files changed, 298 insertions(+), 31 deletions(-) diff --git a/decisions.md b/decisions.md index 584782e7c..71cfc5eef 100644 --- a/decisions.md +++ b/decisions.md @@ -4798,10 +4798,14 @@ is unchanged. Does not change D95, D99, or D100. **Decision.** The converter contract takes a **source handle** — a read-only capability on one immutable, already-hashed source — in place of the whole file as a `bytes` value (amends D65's `convert(bytes, mime, hints)` and, through it, -D38/D57). The handle offers three reads and deliberately no fourth: **stream** -the source in bounded chunks, **read a half-open byte range** `[start, end)`, -or **materialise** it to a temporary local file. There is no operation that -returns the entire source as bytes. `ObjectStorePort` grows the matching +D38/D57). The handle offers four reads, every one of them bounded: **stream** the source +in bounded chunks, **read a half-open byte range** `[start, end)`, +**materialise** it to a temporary local file, or **read it bounded** into +memory after naming the limit. There is no operation that reads a source of +unknown size — the hazard was never bytes in memory, it was bytes in memory +without anyone having decided how many. Reads never come back short: a range +returns exactly what was asked for or raises, because a parser handed a +truncated header cannot tell it from a valid one. `ObjectStorePort` grows the matching `open_stream` and `read_range` so the handle has something to sit on, and the existing `read_bytes` stays for the small text objects every shipped route converts. @@ -4810,10 +4814,10 @@ converts. bytes-first contract the client, the HTTP process, the object store, and the converter each materialise a complete copy of the same file, so two concurrent large ingests can exhaust a worker that would have processed either one alone. -That is an availability property, not an optimisation. The absence of a -whole-file read is the load-bearing part of the decision: a contract that -offers buffering as the convenient option gets buffering, and the failure -surfaces on the first large file in production rather than in review. +That is an availability property, not an optimisation. Requiring every read to +carry a bound is the load-bearing part of the decision: a contract that offers +an unbounded read as the convenient option gets unbounded reads, and the +failure surfaces on the first large file in production rather than in review. **Bounds live in materialisation**, because that is the operation that can exhaust a host. The caller declares what it can afford and an oversized source @@ -4839,5 +4843,7 @@ D57 (Markdown coordinate system), D61 (port boundary), D65 (media routes). scales the blast radius with source size and solves neither resumability nor cancellation. *Hand routes a raw provider object* — rejected: it puts a storage SDK type in the converter contract, which D61 exists to prevent. *Give the -handle a `read_all()` for convenience* — rejected: it becomes the path of least -resistance and reintroduces exactly the failure this decision removes. +handle an unbounded `read_all()` for convenience* — rejected: it becomes the +path of least resistance and reintroduces exactly the failure this decision +removes. `read_bounded(max_bytes=...)` serves the same routes while forcing the +caller to state a limit, which is the property that actually matters. diff --git a/plan/designs/media_design.md b/plan/designs/media_design.md index 0fdfa8f5c..ed9672210 100644 --- a/plan/designs/media_design.md +++ b/plan/designs/media_design.md @@ -134,8 +134,8 @@ take down a worker that would happily have processed either one alone. So a route receives a **source handle** — a read-only capability on one immutable, already-hashed source — and decides for itself how much of it to -bring into memory. The handle offers exactly three ways to read, and -deliberately no fourth: +bring into memory. The handle offers four ways to read, every one of them +bounded: - **stream it** — receive the source in order, one bounded chunk at a time, which is what hashing, copying, and demuxing need; @@ -145,12 +145,25 @@ deliberately no fourth: to remember which end is inclusive; and - **materialise it** — write the source to a temporary local file and hand over the path. Decoders like FFmpeg want a seekable file rather than a - stream, and refusing to provide one would mean reimplementing them. - -There is deliberately **no method that returns the whole source as bytes**. -The absence is the point: a contract that offers whole-file buffering as the -convenient option gets whole-file buffering, and the failure only shows up on -the first large file in production. + stream, and refusing to provide one would mean reimplementing them; and +- **read it bounded** — take the whole source into memory, having first said + how much memory that is allowed to be. A Markdown passthrough has no use for + a stream, and pretending otherwise would push every small route into writing + its own accumulation loop. + +There is deliberately **no method that reads a source of unknown size**. Every +read either bounds itself structurally (a chunk, a range) or requires the +caller to name a limit up front. The distinction matters: the hazard was never +"bytes in memory", it was *bytes in memory without anyone having decided how +many*. A contract that offers an unbounded whole-file read as the convenient +option gets unbounded whole-file reads, and the failure only shows up on the +first large file in production. + +Reads never come back short. A range returns exactly the bytes it was asked +for or it raises — a container parser handed a truncated header cannot tell it +from a valid one, and will produce confident nonsense rather than an error. +That check belongs here rather than in each route, because every route would +otherwise have to remember to write it. Materialisation is where the bounds live, because it is the operation that can actually exhaust a host. Three properties are enforced by the handle rather diff --git a/src/rememberstack/adapters/selfhost/minio.py b/src/rememberstack/adapters/selfhost/minio.py index fba2c5eec..a0d108cfe 100644 --- a/src/rememberstack/adapters/selfhost/minio.py +++ b/src/rememberstack/adapters/selfhost/minio.py @@ -185,16 +185,34 @@ def read_range(self, *, key: ObjectKey, start: int, end: int) -> bytes: raise SourceRangeError( f"range [{start}, {end}) is empty, reversed, or negative" ) - response = self._client.get_object( - Bucket=self._bucket, - Key=_validated_key(key=key), - Range=f"bytes={start}-{end - 1}", - ) + try: + response = self._client.get_object( + Bucket=self._bucket, + Key=_validated_key(key=key), + Range=f"bytes={start}-{end - 1}", + ) + except ClientError as error: + # A range past the end is a 416 from S3. Letting botocore's + # exception escape would put a provider type in the port contract + # that D61 exists to keep provider-free. + code = error.response.get("Error", {}).get("Code", "") + status = error.response.get("ResponseMetadata", {}).get("HTTPStatusCode") + if code in {"InvalidRange", "416"} or status == 416: + raise SourceRangeError( + f"range [{start}, {end}) is outside {key.root!r}" + ) from error + raise body = response["Body"] try: - return body.read() + content = body.read() finally: body.close() + if len(content) != end - start: + raise SourceRangeError( + f"range [{start}, {end}) of {key.root!r} returned {len(content)} " + f"bytes, not the {end - start} requested" + ) + return content def write_bytes( self, *, key: ObjectKey, content: bytes, storage_class: str | None = None diff --git a/src/rememberstack/adapters/selfhost/object_store.py b/src/rememberstack/adapters/selfhost/object_store.py index 7c4a06a85..91a87b1c1 100644 --- a/src/rememberstack/adapters/selfhost/object_store.py +++ b/src/rememberstack/adapters/selfhost/object_store.py @@ -39,7 +39,13 @@ def read_range(self, *, key: ObjectKey, start: int, end: int) -> bytes: ) with self._path_for(key=key).open(mode="rb") as handle: handle.seek(start) - return handle.read(end - start) + content = handle.read(end - start) + if len(content) != end - start: + raise SourceRangeError( + f"range [{start}, {end}) of {key.root!r} returned {len(content)} " + f"bytes, not the {end - start} requested" + ) + return content def write_bytes( self, *, key: ObjectKey, content: bytes, storage_class: str | None = None diff --git a/src/rememberstack/adapters/selfhost/source_handle.py b/src/rememberstack/adapters/selfhost/source_handle.py index a78bba759..ab2cead17 100644 --- a/src/rememberstack/adapters/selfhost/source_handle.py +++ b/src/rememberstack/adapters/selfhost/source_handle.py @@ -10,6 +10,7 @@ from contextlib import contextmanager import hashlib from pathlib import Path +import shutil import tempfile from rememberstack.model import SourceHashMismatchError @@ -68,9 +69,34 @@ def read_range(self, *, start: int, end: int) -> bytes: f"range [{start}, {end}) extends past the recorded source size " f"{self._identity.byte_size}" ) - return self._store.read_range( + content = self._store.read_range( key=self._identity.object_key, start=start, end=end ) + if len(content) != end - start: + raise SourceRangeError( + f"range [{start}, {end}) of {self._identity.object_key.root!r} " + f"returned {len(content)} bytes, not the {end - start} requested; " + "the stored object is shorter than its recorded size" + ) + return content + + def read_bounded(self, *, max_bytes: int) -> bytes: + """Read the whole source into memory, refusing anything over the bound. + + Small routes legitimately want every byte — a Markdown passthrough has + no use for a stream. This exists so they are not driven to write + ``b"".join(handle.open_stream())``, which is an unbounded read wearing + a bounded read's clothes. Naming the bound is the whole point: there is + still no way to ask for a source of unknown size. + """ + if max_bytes < 0: + raise ValueError(f"max_bytes must be non-negative, got {max_bytes}") + if self._identity.byte_size > max_bytes: + raise SourceTooLargeError( + f"source {self._identity.object_key.root!r} is " + f"{self._identity.byte_size} bytes, over the {max_bytes} accepted" + ) + return self._store.read_bytes(key=self._identity.object_key) @contextmanager def materialize_seekable(self, *, max_bytes: int) -> Iterator[Path]: @@ -86,8 +112,8 @@ def materialize_seekable(self, *, max_bytes: int) -> Iterator[Path]: converted into a plausible-looking short document; and * the file is removed on the way out whether the body raised or not. """ - if max_bytes <= 0: - raise SourceTooLargeError(f"max_bytes must be positive, got {max_bytes}") + if max_bytes < 0: + raise ValueError(f"max_bytes must be non-negative, got {max_bytes}") if self._identity.byte_size > max_bytes: raise SourceTooLargeError( f"source {self._identity.object_key.root!r} is " @@ -99,7 +125,10 @@ def materialize_seekable(self, *, max_bytes: int) -> Iterator[Path]: prefix="rememberstack-source-", dir=None if self._temp_root is None else str(self._temp_root), ) - path = Path(directory) / "source" + # Demuxers routinely sniff the container from the extension before + # falling back to probing, so carrying the source's own suffix across + # costs nothing and avoids a class of "decoder cannot open it" bug. + path = Path(directory) / f"source{Path(self._identity.object_key.root).suffix}" try: digest = hashlib.sha256() written = 0 @@ -122,5 +151,8 @@ def materialize_seekable(self, *, max_bytes: int) -> Iterator[Path]: ) yield path finally: - path.unlink(missing_ok=True) - Path(directory).rmdir() + # rmtree, not unlink plus rmdir: a decoder handed a path routinely + # writes beside it (indexes, sidecars, split tracks), and a + # directory-not-empty error here would both leak the directory and + # mask whatever the route was already raising. + shutil.rmtree(directory, ignore_errors=True) diff --git a/src/rememberstack/ports/source_handle.py b/src/rememberstack/ports/source_handle.py index 97752c3e2..ce4540b7a 100644 --- a/src/rememberstack/ports/source_handle.py +++ b/src/rememberstack/ports/source_handle.py @@ -39,6 +39,20 @@ def read_range(self, *, start: int, end: int) -> bytes: Half-open to match every other interval in the system (D65 locators), so a caller never has to remember which end is inclusive. + + Returns exactly ``end - start`` bytes or raises. A short read is never + returned: a container parser that receives a truncated header cannot + tell it from a valid one and will produce confident nonsense. + """ + ... + + def read_bounded(self, *, max_bytes: int) -> bytes: + """Read the whole source into memory, refusing anything over the bound. + + For routes that legitimately want every byte — a Markdown passthrough + has no use for a stream. The bound must be named, so this is still not + a way to read a source of unknown size; it exists so small routes are + not driven to write their own unbounded ``join`` over `open_stream`. """ ... diff --git a/src/tests/adapters/test_minio_store.py b/src/tests/adapters/test_minio_store.py index 3ddf55396..44c9701b9 100644 --- a/src/tests/adapters/test_minio_store.py +++ b/src/tests/adapters/test_minio_store.py @@ -12,6 +12,7 @@ from rememberstack.model import ObjectAlreadyExistsError from rememberstack.model import ObjectKey from rememberstack.model import ObjectKeyEscapesRootError +from rememberstack.model import SourceRangeError class _Body: @@ -163,3 +164,64 @@ def _client_error(*, code: str, operation: str) -> ClientError: error_response={"Error": {"Code": code, "Message": code}}, operation_name=operation, ) + + +def test_stream_reassembles_and_releases_the_connection() -> None: + """Chunked reads cover the object and still close the HTTP body.""" + client = _MemoryS3() + store = MinIOObjectStore(bucket="raw", client=client) + store.ensure_bucket() + content = bytes(range(256)) * 8 + store.write_bytes(key=ObjectKey("clip.mp4"), content=content) + + chunks = list(store.open_stream(key=ObjectKey("clip.mp4"), chunk_bytes=100)) + + assert b"".join(chunks) == content + assert max(len(chunk) for chunk in chunks) <= 100 + assert client.last_body is not None and client.last_body.closed + + +def test_range_translates_half_open_to_the_inclusive_wire_form() -> None: + """``[2, 5)`` must reach S3 as ``bytes=2-4``, not ``bytes=2-5``.""" + client = _MemoryS3() + store = MinIOObjectStore(bucket="raw", client=client) + store.ensure_bucket() + store.write_bytes(key=ObjectKey("clip.mp4"), content=b"0123456789") + + assert store.read_range(key=ObjectKey("clip.mp4"), start=2, end=5) == b"234" + assert client.last_body is not None and client.last_body.closed + + +def test_range_past_the_object_end_raises_a_domain_error() -> None: + """A short provider response must not reach the caller as valid bytes.""" + client = _MemoryS3() + store = MinIOObjectStore(bucket="raw", client=client) + store.ensure_bucket() + store.write_bytes(key=ObjectKey("clip.mp4"), content=b"01234") + + with pytest.raises(SourceRangeError): + store.read_range(key=ObjectKey("clip.mp4"), start=2, end=99) + + +@pytest.mark.parametrize( + ("start", "end"), [(5, 5), (5, 2), (-1, 4)], ids=["empty", "reversed", "negative"] +) +def test_degenerate_ranges_are_refused_before_any_request(start: int, end: int) -> None: + """A malformed range is a caller bug and never becomes a wire request.""" + client = _MemoryS3() + store = MinIOObjectStore(bucket="raw", client=client) + store.ensure_bucket() + store.write_bytes(key=ObjectKey("clip.mp4"), content=b"01234") + client.last_body = None + + with pytest.raises(SourceRangeError): + store.read_range(key=ObjectKey("clip.mp4"), start=start, end=end) + assert client.last_body is None, "a refused range must not reach the provider" + + +def test_nonpositive_chunk_size_is_refused() -> None: + """A zero chunk would never advance; a negative one is meaningless.""" + store = MinIOObjectStore(bucket="raw", client=_MemoryS3()) + + with pytest.raises(SourceRangeError): + list(store.open_stream(key=ObjectKey("clip.mp4"), chunk_bytes=0)) diff --git a/src/tests/adapters/test_source_handle.py b/src/tests/adapters/test_source_handle.py index b60e4309c..c4537b270 100644 --- a/src/tests/adapters/test_source_handle.py +++ b/src/tests/adapters/test_source_handle.py @@ -155,3 +155,119 @@ def test_nonpositive_chunk_size_is_refused(tmp_path: Path) -> None: with pytest.raises(SourceRangeError): list(handle.open_stream(chunk_bytes=0)) + + +def test_zero_byte_source_materializes_under_a_zero_bound(tmp_path: Path) -> None: + """An empty file is a valid source; `max_bytes=0` is enough room for it.""" + handle = _handle(tmp_path=tmp_path, content=b"") + + with handle.materialize_seekable(max_bytes=0) as path: + assert path.read_bytes() == b"" + + +def test_negative_bound_is_a_caller_bug_not_an_oversized_source(tmp_path: Path) -> None: + """A negative bound is nonsense, and must not be reported as size refusal.""" + handle = _handle(tmp_path=tmp_path, content=b"data") + + with pytest.raises(ValueError): + with handle.materialize_seekable(max_bytes=-1): + pytest.fail("a negative bound must never materialize anything") + + +def test_materialized_name_keeps_the_source_suffix(tmp_path: Path) -> None: + """Demuxers sniff the container from the extension before probing.""" + handle = _handle(tmp_path=tmp_path, content=b"video") + + with handle.materialize_seekable(max_bytes=1024) as path: + assert path.suffix == ".mp4" + + +def test_cleanup_survives_files_written_beside_the_source(tmp_path: Path) -> None: + """A decoder writing an index or sidecar must not break teardown. + + Removing only the source file and then the directory would raise + "directory not empty" here, leaking the directory and masking whatever the + route itself was raising. + """ + handle = _handle(tmp_path=tmp_path, content=b"video") + + with handle.materialize_seekable(max_bytes=1024) as path: + (path.parent / "sidecar.idx").write_bytes(b"index") + directory = path.parent + + assert not directory.exists() + + +def test_materialize_without_a_temp_root_uses_the_platform_default( + tmp_path: Path, +) -> None: + """A deployment that has not sized a work volume still gets cleanup.""" + store = LocalFSObjectStore(root=tmp_path / "objects") + store.write_bytes(key=_KEY, content=b"payload") + handle = ObjectSourceHandle( + store=store, + identity=SourceIdentity( + object_key=_KEY, + content_hash=hashlib.sha256(b"payload").hexdigest(), + byte_size=len(b"payload"), + mime="video/mp4", + ), + ) + + with handle.materialize_seekable(max_bytes=1024) as path: + assert path.read_bytes() == b"payload" + directory = path.parent + + assert not directory.exists() + + +def test_truncated_object_fails_the_range_read(tmp_path: Path) -> None: + """A stored object shorter than its record must not yield a short read. + + A container parser cannot distinguish a truncated header from a valid one, + so it would produce confident nonsense rather than an error. + """ + store = LocalFSObjectStore(root=tmp_path / "objects") + store.write_bytes(key=_KEY, content=b"short") + handle = ObjectSourceHandle( + store=store, + identity=SourceIdentity( + object_key=_KEY, + content_hash=hashlib.sha256(b"short").hexdigest(), + byte_size=100, + mime="video/mp4", + ), + temp_root=tmp_path / "work", + ) + + with pytest.raises(SourceRangeError): + handle.read_range(start=0, end=50) + + +def test_read_bounded_returns_everything_under_the_bound(tmp_path: Path) -> None: + """Small routes get every byte without writing their own unbounded join.""" + handle = _handle(tmp_path=tmp_path, content=b"# heading\n") + + assert handle.read_bounded(max_bytes=1024) == b"# heading\n" + + +def test_read_bounded_refuses_an_oversized_source(tmp_path: Path) -> None: + """The bound is the point: this is not a whole-file read in disguise.""" + handle = _handle(tmp_path=tmp_path, content=b"x" * 5000) + + with pytest.raises(SourceTooLargeError): + handle.read_bounded(max_bytes=100) + + +def test_concurrent_materializations_do_not_share_a_directory(tmp_path: Path) -> None: + """Two live materializations of one handle must not collide or co-delete.""" + handle = _handle(tmp_path=tmp_path, content=b"payload") + + with handle.materialize_seekable(max_bytes=1024) as first: + with handle.materialize_seekable(max_bytes=1024) as second: + assert first != second + assert first.parent != second.parent + assert first.read_bytes() == second.read_bytes() == b"payload" + inner = second.parent + assert not inner.exists() + assert first.exists(), "the outer materialization must survive the inner" From 24472a1cf393a26ffaa8d17a8371916e284a4197 Mon Sep 17 00:00:00 2001 From: Jiri Puc Date: Tue, 1 Sep 2026 02:27:24 +0200 Subject: [PATCH 3/3] =?UTF-8?q?fix(media):=20second=20review=20round=20?= =?UTF-8?q?=E2=80=94=20stream=20lifetime,=20read=20ceiling,=20size=20truth?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Second independent review (Codex) converged with the first on both BLOCKs already fixed, and found five more. All addressed. The sharpest one was aimed at my own claim. D104 said there is no whole-file read, but read_range(0, byte_size) is exactly that with extra steps, and nothing capped a range or a chunk. The claim was ergonomic friction dressed as an availability invariant. A handle now takes an optional max_read_bytes ceiling that refuses any single range, chunk, or bounded read above it, and the design says plainly that without a ceiling the property is friction, not a guarantee. Overclaiming in a decision doc is worse than the gap it hid. open_stream returned a bare Iterator, so a caller that stopped early held a file descriptor or an HTTP body until garbage collection. GC timing is not a resource-lifetime contract and enough abandoned reads exhaust the descriptor limit or the connection pool. It is now a context manager at every level -- port, both adapters, handle, and the three fakes -- released when the block exits. The streaming guard fired only after a full default chunk had moved: a 100-byte allowance could pull a megabyte before noticing. The read is now sized to the remaining allowance plus one byte, the extra byte being what makes "over" detectable at all -- a read that exactly fills the allowance is indistinguishable from one that stops there. Materialization verified the hash but never the length, so bytes that hash correctly could still be described by a wrong size: a source recorded as 10 bytes materialized happily at 5000 while a range read past 10 was refused -- two access paths disagreeing about one source. Now a typed SourceSizeMismatchError, distinct from the hash error because they catch different lies. SourceIdentity carried a mime field while the converter contract already takes mime as its own argument. Two authorities that can disagree, and a route would have to know which one routing used. Dropped from the identity. Concurrent materializations still have no aggregate temp-disk budget -- two 6 GiB copies each pass their own bound and together need 12 GiB. That is worker-level admission, not a handle concern, so it is documented as a stated boundary in both the design and D104 rather than quietly left. Tests: two asserted less than their names claimed. "Refused before any bytes move" checked only that no file survived, and the understated-size test read its whole 5000-byte source before raising and still passed. Both now use a counting store and assert bytes actually served (0, and <=101). Added coverage for the ceiling, the size mismatch, and releasing a stream abandoned after one chunk. Validation: ruff clean; pyright 0 errors; test inventory OK; 1322 passed, 687 skipped. The single failure, test_sdk_pushes_lineage_metadata_to_e0, reproduces on the base commit without these changes -- mimetypes on this machine does not resolve .md. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01XTv3baR8R5EgEGFePkQMjK --- decisions.md | 19 ++- plan/designs/media_design.md | 27 ++++ src/rememberstack/adapters/selfhost/minio.py | 15 +- .../adapters/selfhost/object_store.py | 15 +- .../adapters/selfhost/source_handle.py | 92 +++++++++-- src/rememberstack/model/__init__.py | 2 + src/rememberstack/model/source_handle.py | 22 ++- src/rememberstack/ports/object_store.py | 17 +- src/rememberstack/ports/source_handle.py | 12 +- src/tests/adapters/test_minio_store.py | 6 +- src/tests/adapters/test_source_handle.py | 146 ++++++++++++++++-- .../spine/test_knowledge_control_plane.py | 15 +- .../test_port_inventory_and_conformance.py | 22 +-- src/tests/workers/test_ingest_admission.py | 15 +- 14 files changed, 357 insertions(+), 68 deletions(-) diff --git a/decisions.md b/decisions.md index 71cfc5eef..bd1cec927 100644 --- a/decisions.md +++ b/decisions.md @@ -4803,9 +4803,17 @@ in bounded chunks, **read a half-open byte range** `[start, end)`, **materialise** it to a temporary local file, or **read it bounded** into memory after naming the limit. There is no operation that reads a source of unknown size — the hazard was never bytes in memory, it was bytes in memory -without anyone having decided how many. Reads never come back short: a range -returns exactly what was asked for or raises, because a parser handed a -truncated header cannot tell it from a valid one. `ObjectStorePort` grows the matching +without anyone having decided how many. Because naming a large limit is still +naming one, a handle may carry a **read ceiling** that refuses any single read +above it; without a ceiling the property is friction rather than a guarantee, +and the decision says so rather than overclaiming. A stream is opened for the +duration of a block and released at its end, since the resource underneath is a +descriptor or a connection and garbage-collection timing is not a lifetime +contract. Reads never come back short: a range returns exactly what was asked +for or raises, because a parser handed a truncated header cannot tell it from a +valid one. Recorded length is verified alongside recorded hash — bytes can hash +correctly and still be described by a wrong size, which would let two access +paths disagree about one source. `ObjectStorePort` grows the matching `open_stream` and `read_range` so the handle has something to sit on, and the existing `read_bytes` stays for the small text objects every shipped route converts. @@ -4832,6 +4840,11 @@ source that exceeds both the declaration and the accepted bound. The temporary file's lifetime belongs to the handle, so a route that fails mid-decode cannot leak the file it asked for. +**Not solved here.** Nothing reserves worker-wide temporary disk, so two +concurrent materialisations of one source each pass their own bound and can +together exceed the volume. Aggregate admission belongs to whatever schedules +the work; this is a stated boundary, not an oversight. + **Consequences.** One contract serves both deployment shapes: a self-host directory tree and a cloud object store present the same three operations, so no route branches on where bytes live. Decoders that require a seekable path diff --git a/plan/designs/media_design.md b/plan/designs/media_design.md index ed9672210..84d533ba6 100644 --- a/plan/designs/media_design.md +++ b/plan/designs/media_design.md @@ -159,12 +159,39 @@ many*. A contract that offers an unbounded whole-file read as the convenient option gets unbounded whole-file reads, and the failure only shows up on the first large file in production. +Naming a limit is not by itself a guarantee, because a caller can name a large +one — asking for the range `[0, size)` is a whole-file read with extra steps. +A handle can therefore carry a **read ceiling**: the largest single range, +chunk, or bounded read it will serve, refused above that regardless of what the +caller asks for. Without a ceiling the property is ergonomic friction; with one +it is enforced. Deployments handling media set it; the small-text routes that +have no size problem do not need it. + +Reading is also a **resource**, not just bytes. Underneath a stream is a file +descriptor or an HTTP connection, so a stream is opened for the duration of a +block and released when the block ends. Returning a bare iterator would leave a +caller who stops early holding that resource until garbage collection, and +collection timing is not a lifetime contract — enough abandoned reads exhaust +the descriptor limit or the connection pool. + Reads never come back short. A range returns exactly the bytes it was asked for or it raises — a container parser handed a truncated header cannot tell it from a valid one, and will produce confident nonsense rather than an error. That check belongs here rather than in each route, because every route would otherwise have to remember to write it. +The recorded **length** is checked as well as the recorded hash, and they catch +different lies. Bytes can hash correctly while being described by a wrong size; +if only the hash were verified, materialisation would happily produce 5,000 +bytes for a source recorded as 10 while a range read past byte 10 was refused — +two access paths disagreeing about the length of one source. + +**Not solved here:** nothing reserves worker-wide temporary disk. Two +concurrent materialisations of one 6 GiB source each pass a 6 GiB bound and +together need 12 GiB. Aggregate admission belongs to whatever schedules the +work; the handle bounds one read at a time, and says so rather than implying +otherwise. + Materialisation is where the bounds live, because it is the operation that can actually exhaust a host. Three properties are enforced by the handle rather than trusted to each route: diff --git a/src/rememberstack/adapters/selfhost/minio.py b/src/rememberstack/adapters/selfhost/minio.py index a0d108cfe..7f7bca450 100644 --- a/src/rememberstack/adapters/selfhost/minio.py +++ b/src/rememberstack/adapters/selfhost/minio.py @@ -1,6 +1,7 @@ """S3-compatible MinIO object storage for the self-host profile.""" from collections.abc import Iterator +from contextlib import contextmanager from typing import cast from typing import NotRequired from typing import Protocol @@ -158,10 +159,11 @@ def read_bytes(self, *, key: ObjectKey) -> bytes: finally: body.close() + @contextmanager def open_stream( self, *, key: ObjectKey, chunk_bytes: int = 1024 * 1024 - ) -> Iterator[bytes]: - """Yield one object in order, holding at most one chunk at a time.""" + ) -> Iterator[Iterator[bytes]]: + """Open an ordered chunked read, releasing the HTTP body on exit.""" if chunk_bytes <= 0: raise SourceRangeError(f"chunk_bytes must be positive, got {chunk_bytes}") response = self._client.get_object( @@ -169,8 +171,13 @@ def open_stream( ) body = response["Body"] try: - while chunk := body.read(chunk_bytes): - yield chunk + + def chunks() -> Iterator[bytes]: + """Yield successive reads until the body is exhausted.""" + while chunk := body.read(chunk_bytes): + yield chunk + + yield chunks() finally: body.close() diff --git a/src/rememberstack/adapters/selfhost/object_store.py b/src/rememberstack/adapters/selfhost/object_store.py index 91a87b1c1..99bc85434 100644 --- a/src/rememberstack/adapters/selfhost/object_store.py +++ b/src/rememberstack/adapters/selfhost/object_store.py @@ -1,6 +1,7 @@ """Local-filesystem object store adapter: immutable bytes under one root (D61/D62).""" from collections.abc import Iterator +from contextlib import contextmanager from pathlib import Path from rememberstack.model import ObjectAlreadyExistsError @@ -21,15 +22,21 @@ def read_bytes(self, *, key: ObjectKey) -> bytes: """Read all bytes stored under an existing object key.""" return self._path_for(key=key).read_bytes() + @contextmanager def open_stream( self, *, key: ObjectKey, chunk_bytes: int = 1024 * 1024 - ) -> Iterator[bytes]: - """Yield one object in order, holding at most one chunk at a time.""" + ) -> Iterator[Iterator[bytes]]: + """Open an ordered chunked read, closing the file when the block exits.""" if chunk_bytes <= 0: raise SourceRangeError(f"chunk_bytes must be positive, got {chunk_bytes}") with self._path_for(key=key).open(mode="rb") as handle: - while chunk := handle.read(chunk_bytes): - yield chunk + + def chunks() -> Iterator[bytes]: + """Yield successive fixed-size reads until the file is spent.""" + while chunk := handle.read(chunk_bytes): + yield chunk + + yield chunks() def read_range(self, *, key: ObjectKey, start: int, end: int) -> bytes: """Read the half-open byte interval ``[start, end)`` of one object.""" diff --git a/src/rememberstack/adapters/selfhost/source_handle.py b/src/rememberstack/adapters/selfhost/source_handle.py index ab2cead17..431382d6b 100644 --- a/src/rememberstack/adapters/selfhost/source_handle.py +++ b/src/rememberstack/adapters/selfhost/source_handle.py @@ -16,6 +16,7 @@ from rememberstack.model import SourceHashMismatchError from rememberstack.model import SourceIdentity from rememberstack.model import SourceRangeError +from rememberstack.model import SourceSizeMismatchError from rememberstack.model import SourceTooLargeError from rememberstack.ports import ObjectStorePort @@ -31,27 +32,60 @@ def __init__( store: ObjectStorePort, identity: SourceIdentity, temp_root: Path | None = None, + max_read_bytes: int | None = None, ) -> None: """Bind the handle to one source and the store that holds its bytes. `temp_root` is where `materialize_seekable` writes. Leaving it None - uses the platform temporary directory; a deployment that bounds - worker disk points it at the volume it actually sized. + uses the platform temporary directory; a deployment that bounds worker + disk points it at the volume it actually sized. + + `max_read_bytes` is the ceiling on any single in-memory read — the + largest range or chunk this handle will serve. Without it, asking for + the range ``[0, byte_size)`` is a whole-file read with extra steps, so + the "no unbounded read" property is only ergonomic friction. With it, + the property is enforced. Left None the handle keeps the friction and + not the guarantee, which is the right default for the small-text routes + that have no size problem. + + **Concurrency.** Each call reads independently; nothing here reserves + worker-wide temporary disk, so two concurrent materializations of a + 6 GiB source can both pass a 6 GiB bound and together need 12 GiB. + Aggregate admission belongs to whatever schedules the work, and is a + documented non-goal of this port rather than an oversight. """ + if max_read_bytes is not None and max_read_bytes <= 0: + raise ValueError( + f"max_read_bytes must be positive when set, got {max_read_bytes}" + ) self._store = store self._identity = identity self._temp_root = temp_root + self._max_read_bytes = max_read_bytes @property def identity(self) -> SourceIdentity: """The source's recorded key, content hash, size, and declared type.""" return self._identity - def open_stream(self, *, chunk_bytes: int = _HASH_CHUNK_BYTES) -> Iterator[bytes]: - """Yield the source in order, holding at most one chunk at a time.""" - return self._store.open_stream( + @contextmanager + def open_stream( + self, *, chunk_bytes: int = _HASH_CHUNK_BYTES + ) -> Iterator[Iterator[bytes]]: + """Open an ordered chunked read, released when the block exits.""" + self._check_read_ceiling(requested=chunk_bytes, what="chunk") + with self._store.open_stream( key=self._identity.object_key, chunk_bytes=chunk_bytes - ) + ) as chunks: + yield chunks + + def _check_read_ceiling(self, *, requested: int, what: str) -> None: + """Refuse a single read larger than this handle's configured ceiling.""" + if self._max_read_bytes is not None and requested > self._max_read_bytes: + raise SourceTooLargeError( + f"{what} of {requested} bytes exceeds this handle's " + f"{self._max_read_bytes}-byte read ceiling" + ) def read_range(self, *, start: int, end: int) -> bytes: """Read the half-open byte interval ``[start, end)`` of the source. @@ -69,6 +103,7 @@ def read_range(self, *, start: int, end: int) -> bytes: f"range [{start}, {end}) extends past the recorded source size " f"{self._identity.byte_size}" ) + self._check_read_ceiling(requested=end - start, what="range") content = self._store.read_range( key=self._identity.object_key, start=start, end=end ) @@ -96,7 +131,14 @@ def read_bounded(self, *, max_bytes: int) -> bytes: f"source {self._identity.object_key.root!r} is " f"{self._identity.byte_size} bytes, over the {max_bytes} accepted" ) - return self._store.read_bytes(key=self._identity.object_key) + self._check_read_ceiling(requested=self._identity.byte_size, what="read") + content = self._store.read_bytes(key=self._identity.object_key) + if len(content) != self._identity.byte_size: + raise SourceSizeMismatchError( + f"source {self._identity.object_key.root!r} holds {len(content)} " + f"bytes, not the {self._identity.byte_size} recorded" + ) + return content @contextmanager def materialize_seekable(self, *, max_bytes: int) -> Iterator[Path]: @@ -132,17 +174,33 @@ def materialize_seekable(self, *, max_bytes: int) -> Iterator[Path]: try: digest = hashlib.sha256() written = 0 + # Read at most the remaining allowance plus one byte, so an object + # whose real length exceeds its recorded size cannot move a whole + # default chunk before the guard notices. The extra byte is what + # makes "too large" detectable at all: a read that exactly fills + # the allowance is indistinguishable from one that stops there. + chunk_bytes = min(_HASH_CHUNK_BYTES, max_bytes + 1) with path.open(mode="wb") as handle: - for chunk in self.open_stream(): - written += len(chunk) - if written > max_bytes: - raise SourceTooLargeError( - f"source {self._identity.object_key.root!r} exceeded the " - f"{max_bytes} accepted while streaming; the recorded size " - f"{self._identity.byte_size} was wrong" - ) - digest.update(chunk) - handle.write(chunk) + with self.open_stream(chunk_bytes=chunk_bytes) as chunks: + for chunk in chunks: + written += len(chunk) + if written > max_bytes: + raise SourceTooLargeError( + f"source {self._identity.object_key.root!r} exceeded " + f"the {max_bytes} accepted while streaming; the " + f"recorded size {self._identity.byte_size} was wrong" + ) + digest.update(chunk) + handle.write(chunk) + if written != self._identity.byte_size: + # Caught separately from the hash: bytes can hash correctly and + # still be described by a wrong length, which would let + # materialization and range reads disagree about how long the + # same source is. + raise SourceSizeMismatchError( + f"source {self._identity.object_key.root!r} holds {written} " + f"bytes, not the {self._identity.byte_size} recorded" + ) if digest.hexdigest() != self._identity.content_hash: raise SourceHashMismatchError( f"source {self._identity.object_key.root!r} hashed to " diff --git a/src/rememberstack/model/__init__.py b/src/rememberstack/model/__init__.py index f243b2bf5..0ba43b2bb 100644 --- a/src/rememberstack/model/__init__.py +++ b/src/rememberstack/model/__init__.py @@ -349,6 +349,7 @@ from rememberstack.model.source_handle import SourceHashMismatchError from rememberstack.model.source_handle import SourceIdentity from rememberstack.model.source_handle import SourceRangeError +from rememberstack.model.source_handle import SourceSizeMismatchError from rememberstack.model.source_handle import SourceTooLargeError from rememberstack.model.spend_lease import SpendLeaseRefused from rememberstack.model.spend_lease import SpendLeaseUnavailable @@ -506,6 +507,7 @@ "SourceHashMismatchError", "SourceIdentity", "SourceRangeError", + "SourceSizeMismatchError", "SourceTooLargeError", "ObservationAssertion", "ObservationCandidate", diff --git a/src/rememberstack/model/source_handle.py b/src/rememberstack/model/source_handle.py index 934baeb93..acfec9ade 100644 --- a/src/rememberstack/model/source_handle.py +++ b/src/rememberstack/model/source_handle.py @@ -20,9 +20,13 @@ class SourceIdentity(BaseModel): """What a converter may know about a source without reading it. The hash and size are the E0 facts recorded when the raw object was - written, not values a converter or a caller supplies. `mime` is the - caller's declared type and stays an untrusted hint: routing and rating - belong to a structural probe of the actual bytes, never to this field. + written, not values a converter or a caller supplies. + + Deliberately no MIME field. The converter contract already carries the + declared type as its own argument (`convert(source, mime, hints)`), and + holding it in two places invites the two copies to disagree — a route + would then have to know which one routing used. One authority, and it is + not this object. """ model_config = ConfigDict(frozen=True, extra="forbid") @@ -30,7 +34,6 @@ class SourceIdentity(BaseModel): object_key: ObjectKey content_hash: Sha256Hex byte_size: int = Field(ge=0) - mime: str = Field(min_length=1) class SourceTooLargeError(Exception): @@ -51,4 +54,13 @@ class SourceHashMismatchError(Exception): class SourceRangeError(Exception): - """A byte range that is empty, reversed, or outside the source.""" + """A byte range that is empty, reversed, outside the source, or short.""" + + +class SourceSizeMismatchError(Exception): + """The stored object's real length is not the length recorded for it. + + Distinct from a hash mismatch: the bytes may hash correctly and still be + described by a wrong size, which would let materialization and range reads + disagree about how long the same source is. + """ diff --git a/src/rememberstack/ports/object_store.py b/src/rememberstack/ports/object_store.py index d3676fd55..8e0b6bd69 100644 --- a/src/rememberstack/ports/object_store.py +++ b/src/rememberstack/ports/object_store.py @@ -1,6 +1,7 @@ """D61 byte/object-key seam for immutable raw inputs, artifacts, and snapshots.""" from collections.abc import Iterator +from contextlib import AbstractContextManager from typing import Protocol from typing import runtime_checkable @@ -22,12 +23,22 @@ def read_bytes(self, *, key: ObjectKey) -> bytes: def open_stream( self, *, key: ObjectKey, chunk_bytes: int = 1024 * 1024 - ) -> Iterator[bytes]: - """Yield one object in order, holding at most one chunk at a time.""" + ) -> AbstractContextManager[Iterator[bytes]]: + """Open an ordered chunked read, released when the block exits. + + A context manager rather than a bare iterator because the underlying + resource is a file descriptor or an HTTP body. A caller that stops + iterating early would otherwise hold it until garbage collection, and + collection timing is not a resource-lifetime contract — enough + abandoned reads exhaust the descriptor limit or the connection pool. + """ ... def read_range(self, *, key: ObjectKey, start: int, end: int) -> bytes: - """Read the half-open byte interval ``[start, end)`` of one object.""" + """Read the half-open byte interval ``[start, end)`` of one object. + + Returns exactly ``end - start`` bytes or raises `SourceRangeError`. + """ ... def write_bytes( diff --git a/src/rememberstack/ports/source_handle.py b/src/rememberstack/ports/source_handle.py index ce4540b7a..4b2680852 100644 --- a/src/rememberstack/ports/source_handle.py +++ b/src/rememberstack/ports/source_handle.py @@ -30,8 +30,16 @@ def identity(self) -> SourceIdentity: """The source's recorded key, content hash, size, and declared type.""" ... - def open_stream(self, *, chunk_bytes: int = 1024 * 1024) -> Iterator[bytes]: - """Yield the source in order, holding at most one chunk at a time.""" + def open_stream( + self, *, chunk_bytes: int = 1024 * 1024 + ) -> AbstractContextManager[Iterator[bytes]]: + """Open an ordered chunked read, released when the block exits. + + A context manager rather than a bare iterator: the resource underneath + is a file descriptor or an HTTP body, and a caller that stops early + would otherwise hold it until garbage collection. Collection timing is + not a resource-lifetime contract. + """ ... def read_range(self, *, start: int, end: int) -> bytes: diff --git a/src/tests/adapters/test_minio_store.py b/src/tests/adapters/test_minio_store.py index 44c9701b9..e8e1e30e5 100644 --- a/src/tests/adapters/test_minio_store.py +++ b/src/tests/adapters/test_minio_store.py @@ -174,7 +174,8 @@ def test_stream_reassembles_and_releases_the_connection() -> None: content = bytes(range(256)) * 8 store.write_bytes(key=ObjectKey("clip.mp4"), content=content) - chunks = list(store.open_stream(key=ObjectKey("clip.mp4"), chunk_bytes=100)) + with store.open_stream(key=ObjectKey("clip.mp4"), chunk_bytes=100) as parts: + chunks = list(parts) assert b"".join(chunks) == content assert max(len(chunk) for chunk in chunks) <= 100 @@ -224,4 +225,5 @@ def test_nonpositive_chunk_size_is_refused() -> None: store = MinIOObjectStore(bucket="raw", client=_MemoryS3()) with pytest.raises(SourceRangeError): - list(store.open_stream(key=ObjectKey("clip.mp4"), chunk_bytes=0)) + with store.open_stream(key=ObjectKey("clip.mp4"), chunk_bytes=0): + pytest.fail("a non-positive chunk must be refused before any read") diff --git a/src/tests/adapters/test_source_handle.py b/src/tests/adapters/test_source_handle.py index c4537b270..bb99fd988 100644 --- a/src/tests/adapters/test_source_handle.py +++ b/src/tests/adapters/test_source_handle.py @@ -1,5 +1,7 @@ """Bounded source reads: streaming, ranges, size refusal, and hash verification.""" +from collections.abc import Iterator +from contextlib import contextmanager import hashlib from pathlib import Path @@ -11,11 +13,36 @@ from rememberstack.model import SourceHashMismatchError from rememberstack.model import SourceIdentity from rememberstack.model import SourceRangeError +from rememberstack.model import SourceSizeMismatchError from rememberstack.model import SourceTooLargeError _KEY = ObjectKey("raw/recording.mp4") +class _CountingStore(LocalFSObjectStore): + """A local store that records how many bytes it actually handed out.""" + + def __init__(self, *, root: Path) -> None: + """Start with an empty byte counter over the given root.""" + super().__init__(root=root) + self.bytes_served = 0 + + @contextmanager + def open_stream( + self, *, key: ObjectKey, chunk_bytes: int = 1024 * 1024 + ) -> Iterator[Iterator[bytes]]: + """Count every byte yielded so a test can assert what actually moved.""" + with super().open_stream(key=key, chunk_bytes=chunk_bytes) as parts: + + def counted() -> Iterator[bytes]: + """Pass chunks through, adding each to the running total.""" + for chunk in parts: + self.bytes_served += len(chunk) + yield chunk + + yield counted() + + def _handle( *, tmp_path: Path, content: bytes, declared_size: int | None = None ) -> ObjectSourceHandle: @@ -28,7 +55,6 @@ def _handle( object_key=_KEY, content_hash=hashlib.sha256(content).hexdigest(), byte_size=len(content) if declared_size is None else declared_size, - mime="video/mp4", ), temp_root=tmp_path / "work", ) @@ -41,7 +67,8 @@ def test_stream_reassembles_the_source_without_whole_file_chunks( content = bytes(range(256)) * 40 handle = _handle(tmp_path=tmp_path, content=content) - chunks = list(handle.open_stream(chunk_bytes=1024)) + with handle.open_stream(chunk_bytes=1024) as parts: + chunks = list(parts) assert b"".join(chunks) == content assert max(len(chunk) for chunk in chunks) <= 1024 @@ -109,25 +136,58 @@ def test_materialize_removes_the_file_even_when_the_caller_raises( def test_oversized_source_is_refused_before_any_bytes_move(tmp_path: Path) -> None: - """The bound is checked against the recorded size, so refusal costs nothing.""" - handle = _handle(tmp_path=tmp_path, content=b"x" * 5000) + """The bound is checked against the recorded size, so refusal costs nothing. + + Asserted by counting what the store was asked for: the point is not merely + that no file survives, but that the store was never touched at all. + """ + store = _CountingStore(root=tmp_path / "objects") + store.write_bytes(key=_KEY, content=b"x" * 5000) + handle = ObjectSourceHandle( + store=store, + identity=SourceIdentity( + object_key=_KEY, + content_hash=hashlib.sha256(b"x" * 5000).hexdigest(), + byte_size=5000, + ), + temp_root=tmp_path / "work", + ) with pytest.raises(SourceTooLargeError): with handle.materialize_seekable(max_bytes=100): pytest.fail("an oversized source must never be materialized") + assert store.bytes_served == 0, "refusal must precede any read" assert not (tmp_path / "work").exists() or not any((tmp_path / "work").iterdir()) def test_understated_size_is_still_caught_while_streaming(tmp_path: Path) -> None: - """A wrong recorded size cannot be used to smuggle past the accepted bound.""" - content = b"y" * 5000 - handle = _handle(tmp_path=tmp_path, content=content, declared_size=10) + """A wrong recorded size cannot be used to smuggle past the accepted bound. + + The guard must also fire *promptly*: reading a full default chunk before + noticing would move a megabyte for a hundred-byte allowance, so the read is + sized to the remaining allowance plus the one byte that makes "over" visible. + """ + store = _CountingStore(root=tmp_path / "objects") + store.write_bytes(key=_KEY, content=b"y" * 5000) + handle = ObjectSourceHandle( + store=store, + identity=SourceIdentity( + object_key=_KEY, + content_hash=hashlib.sha256(b"y" * 5000).hexdigest(), + byte_size=10, + ), + temp_root=tmp_path / "work", + ) with pytest.raises(SourceTooLargeError): with handle.materialize_seekable(max_bytes=100): pytest.fail("the streaming guard must fire when the size was wrong") + assert store.bytes_served <= 101, ( + f"moved {store.bytes_served} bytes for a 100-byte allowance" + ) + def test_corrupted_bytes_fail_the_hash_check(tmp_path: Path) -> None: """Immutable objects cannot legitimately change, so a mismatch is terminal.""" @@ -139,7 +199,6 @@ def test_corrupted_bytes_fail_the_hash_check(tmp_path: Path) -> None: object_key=_KEY, content_hash=hashlib.sha256(b"different-content").hexdigest(), byte_size=len(b"actual-content"), - mime="video/mp4", ), temp_root=tmp_path / "work", ) @@ -154,7 +213,8 @@ def test_nonpositive_chunk_size_is_refused(tmp_path: Path) -> None: handle = _handle(tmp_path=tmp_path, content=b"data") with pytest.raises(SourceRangeError): - list(handle.open_stream(chunk_bytes=0)) + with handle.open_stream(chunk_bytes=0): + pytest.fail("a non-positive chunk must be refused before any read") def test_zero_byte_source_materializes_under_a_zero_bound(tmp_path: Path) -> None: @@ -210,7 +270,6 @@ def test_materialize_without_a_temp_root_uses_the_platform_default( object_key=_KEY, content_hash=hashlib.sha256(b"payload").hexdigest(), byte_size=len(b"payload"), - mime="video/mp4", ), ) @@ -235,7 +294,6 @@ def test_truncated_object_fails_the_range_read(tmp_path: Path) -> None: object_key=_KEY, content_hash=hashlib.sha256(b"short").hexdigest(), byte_size=100, - mime="video/mp4", ), temp_root=tmp_path / "work", ) @@ -271,3 +329,69 @@ def test_concurrent_materializations_do_not_share_a_directory(tmp_path: Path) -> inner = second.parent assert not inner.exists() assert first.exists(), "the outer materialization must survive the inner" + + +def test_read_ceiling_refuses_a_whole_file_range(tmp_path: Path) -> None: + """With a ceiling set, `[0, byte_size)` stops being a whole-file read. + + Without one the "no unbounded read" property is only friction: a caller + can always ask for the entire range. The ceiling turns it into a rule. + """ + store = LocalFSObjectStore(root=tmp_path / "objects") + store.write_bytes(key=_KEY, content=b"z" * 5000) + handle = ObjectSourceHandle( + store=store, + identity=SourceIdentity( + object_key=_KEY, + content_hash=hashlib.sha256(b"z" * 5000).hexdigest(), + byte_size=5000, + ), + temp_root=tmp_path / "work", + max_read_bytes=1024, + ) + + with pytest.raises(SourceTooLargeError): + handle.read_range(start=0, end=5000) + with pytest.raises(SourceTooLargeError): + handle.read_bounded(max_bytes=10_000) + with pytest.raises(SourceTooLargeError): + with handle.open_stream(chunk_bytes=4096): + pytest.fail("a chunk over the ceiling must be refused") + + assert handle.read_range(start=0, end=1024) == b"z" * 1024 + + +def test_recorded_size_must_match_what_was_actually_stored(tmp_path: Path) -> None: + """Bytes can hash correctly and still be described by a wrong length. + + Left unchecked, materialization would succeed with 5000 bytes while a range + read past 10 was refused — two access paths disagreeing about one source. + """ + content = b"w" * 5000 + store = LocalFSObjectStore(root=tmp_path / "objects") + store.write_bytes(key=_KEY, content=content) + handle = ObjectSourceHandle( + store=store, + identity=SourceIdentity( + object_key=_KEY, + content_hash=hashlib.sha256(content).hexdigest(), + byte_size=10, + ), + temp_root=tmp_path / "work", + ) + + with pytest.raises(SourceSizeMismatchError): + with handle.materialize_seekable(max_bytes=10_000): + pytest.fail("a size mismatch must not reach the converter") + + +def test_abandoning_a_stream_still_releases_the_file(tmp_path: Path) -> None: + """Stopping early must release the descriptor at block exit, not at GC.""" + handle = _handle(tmp_path=tmp_path, content=b"a" * 4096) + + with handle.open_stream(chunk_bytes=16) as chunks: + first = next(iter(chunks)) + assert first == b"a" * 16 + + # Exiting the block closed the underlying file; the source stays readable. + assert handle.read_range(start=0, end=4) == b"aaaa" diff --git a/src/tests/spine/test_knowledge_control_plane.py b/src/tests/spine/test_knowledge_control_plane.py index d097ec2c4..215a003b4 100644 --- a/src/tests/spine/test_knowledge_control_plane.py +++ b/src/tests/spine/test_knowledge_control_plane.py @@ -1,6 +1,7 @@ """WP-6.1 acceptance: live K control plane, routing, and exact staleness.""" from collections.abc import Iterator +from contextlib import contextmanager from datetime import datetime from datetime import UTC from decimal import Decimal @@ -552,13 +553,19 @@ def read_bytes(self, *, key: ObjectKey) -> bytes: """Return one previously archived session transcript.""" return self.objects[key.root] + @contextmanager def open_stream( self, *, key: ObjectKey, chunk_bytes: int = 1024 * 1024 - ) -> Iterator[bytes]: - """Yield the stored bytes in fixed-size chunks.""" + ) -> Iterator[Iterator[bytes]]: + """Open an ordered chunked read over the stored bytes.""" content = self.read_bytes(key=key) - for offset in range(0, len(content), chunk_bytes): - yield content[offset : offset + chunk_bytes] + + def chunks() -> Iterator[bytes]: + """Yield successive fixed-size slices of the stored bytes.""" + for offset in range(0, len(content), chunk_bytes): + yield content[offset : offset + chunk_bytes] + + yield chunks() def read_range(self, *, key: ObjectKey, start: int, end: int) -> bytes: """Return the half-open byte interval of the stored bytes.""" diff --git a/src/tests/test_port_inventory_and_conformance.py b/src/tests/test_port_inventory_and_conformance.py index 75546f95e..3d208f469 100644 --- a/src/tests/test_port_inventory_and_conformance.py +++ b/src/tests/test_port_inventory_and_conformance.py @@ -1,6 +1,7 @@ """Inventory tests for D61 substrate seams plus D74 store capabilities.""" from collections.abc import Iterator +from contextlib import contextmanager from datetime import datetime from datetime import timezone from decimal import Decimal @@ -99,13 +100,19 @@ def read_bytes(self, *, key: ObjectKey) -> bytes: """Return the bytes stored under the requested key.""" return self.objects[key.root] + @contextmanager def open_stream( self, *, key: ObjectKey, chunk_bytes: int = 1024 * 1024 - ) -> Iterator[bytes]: - """Yield the stored bytes in fixed-size chunks.""" - content = self.objects[key.root] - for offset in range(0, len(content), chunk_bytes): - yield content[offset : offset + chunk_bytes] + ) -> Iterator[Iterator[bytes]]: + """Open an ordered chunked read over the stored bytes.""" + content = self.read_bytes(key=key) + + def chunks() -> Iterator[bytes]: + """Yield successive fixed-size slices of the stored bytes.""" + for offset in range(0, len(content), chunk_bytes): + yield content[offset : offset + chunk_bytes] + + yield chunks() def read_range(self, *, key: ObjectKey, start: int, end: int) -> bytes: """Return the half-open byte interval of the stored bytes.""" @@ -228,10 +235,7 @@ def announce( _source_handle_assignment: SourceHandlePort = ObjectSourceHandle( store=FakeObjectStore(), identity=SourceIdentity( - object_key=ObjectKey("raw/example"), - content_hash="0" * 64, - byte_size=0, - mime="video/mp4", + object_key=ObjectKey("raw/example"), content_hash="0" * 64, byte_size=0 ), ) _mount_assignment: MountPublisherPort = FakeMountPublisher() diff --git a/src/tests/workers/test_ingest_admission.py b/src/tests/workers/test_ingest_admission.py index d75ccd739..03c504743 100644 --- a/src/tests/workers/test_ingest_admission.py +++ b/src/tests/workers/test_ingest_admission.py @@ -1,6 +1,7 @@ """Fast D74 proof that ingest checks admission before persisting bytes.""" from collections.abc import Iterator +from contextlib import contextmanager from datetime import datetime from datetime import timedelta from datetime import timezone @@ -57,13 +58,19 @@ def __init__(self) -> None: def read_bytes(self, *, key: ObjectKey) -> bytes: raise AssertionError(f"unexpected read of {key.root}") + @contextmanager def open_stream( self, *, key: ObjectKey, chunk_bytes: int = 1024 * 1024 - ) -> Iterator[bytes]: - """Yield the stored bytes in fixed-size chunks.""" + ) -> Iterator[Iterator[bytes]]: + """Open an ordered chunked read over the stored bytes.""" content = self.read_bytes(key=key) - for offset in range(0, len(content), chunk_bytes): - yield content[offset : offset + chunk_bytes] + + def chunks() -> Iterator[bytes]: + """Yield successive fixed-size slices of the stored bytes.""" + for offset in range(0, len(content), chunk_bytes): + yield content[offset : offset + chunk_bytes] + + yield chunks() def read_range(self, *, key: ObjectKey, start: int, end: int) -> bytes: """Return the half-open byte interval of the stored bytes."""