diff --git a/docs/blob-storage-addendum.md b/docs/design-blob-storage.md similarity index 100% rename from docs/blob-storage-addendum.md rename to docs/design-blob-storage.md diff --git a/docs/design.md b/docs/design.md old mode 100644 new mode 100755 index d0db194..7dba759 --- a/docs/design.md +++ b/docs/design.md @@ -315,6 +315,7 @@ revenant status # print each stage's offset / running-or-idl - **Adding new input mid-run.** Explicitly deferred — input is currently treated as a fixed, fully-assembled batch known before the pipeline starts (§1, §7). If this changes later, the fix is a single `input.closed.json` marker (`{"final_seq": N, "closed_at": "..."}`) written explicitly whenever input is done being appended to, which becomes the new base case for §7's recursion; every other stage boundary is unaffected. Two operating modes would fall out of this: batch mode (close input immediately, run to drain) vs. long-running/streaming mode (leave input open, keep enqueuing, only close it to deliberately shut the pipeline down for good). - **Snapshot frequency.** Confirmed: checkpoint after *every* completed item (not batched every N items), since tasks may be long-running and the cost of reprocessing more than one item on crash was judged not worth the savings from less-frequent snapshotting. - **Log rotation / compaction.** Not yet designed. `A.jsonl` grows unboundedly for the life of a run; if a stage's output file needs periodic compaction (e.g. dropping already-fully-consumed lines), note that `seq` numbers — not line-position-in-file — are the resumption unit specifically so this remains possible without breaking downstream consumers. +- **Blob storage garbage collection.** Deferred, same bucket as the log rotation/compaction item above — see `docs/design-blob-storage.md` for the full blob-storage design and the one-hop rule that keeps a future GC pass tractable. - **Retry backoff policy for `RetryableError`.** Max attempts / backoff schedule not yet implemented; current behavior is an unconditional retry with the fixed poll-interval delay already used by the runner. - **Docker bind-mount atomicity verification.** Confirm the actual host filesystem backing the bind mount before relying on `write()`-append and `rename()` atomicity (§6's caveat) — this is load-bearing for the entire crash-safety story. diff --git a/examples/blob_split_merge/__init__.py b/examples/blob_split_merge/__init__.py new file mode 100755 index 0000000..e69de29 diff --git a/examples/blob_split_merge/make_input.py b/examples/blob_split_merge/make_input.py new file mode 100755 index 0000000..69f38b0 --- /dev/null +++ b/examples/blob_split_merge/make_input.py @@ -0,0 +1,47 @@ +""" +Seed state/input.jsonl for the blob_split_merge example. + +This script creates a tiny input stream for the blob split/merge +pipeline by writing one logical record that points to a sample file. +Run from the repo root: + python examples/blob_split_merge/make_input.py +""" + +from __future__ import annotations + +from pathlib import Path + +from revenant.io_utils import atomic_append_lines, make_checkpoint_line, make_record_line + +LINES = [ + "README.md", +] + +def main() -> None: + state_dir = Path("state") + state_dir.mkdir(exist_ok=True) + input_path = state_dir / "input.jsonl" + if input_path.exists(): + input_path.unlink() + + # Create one logical input record for the sample file. + records = [ + make_record_line( + seq=i, + src_seq=i, + parent_seq=i, + payload={"input": text}, + ) + for i, text in enumerate(LINES, start=1) + ] + # Treat the generated input as already committed so downstream + # stages can consume it immediately. The checkpoint line marks the + # latest visible state for tooling that reads the last committed + # record (see docs/design.md, section 5.1). + records.append(make_checkpoint_line(last_consumed_seq=len(LINES), last_emitted_seq=len(LINES))) + atomic_append_lines(input_path, records) + print(f"Wrote {len(LINES)} input records to {input_path}") + + +if __name__ == "__main__": + main() diff --git a/examples/blob_split_merge/pipeline.py b/examples/blob_split_merge/pipeline.py new file mode 100755 index 0000000..60297b6 --- /dev/null +++ b/examples/blob_split_merge/pipeline.py @@ -0,0 +1,64 @@ +""" +Example pipeline: blob split/merge. + +Demonstrates: + - a stateless stage that fans out one input file into many blob-backed + parts (SplitFile: one file -> many part records) + - a stateful stage that reassembles those parts once all fragments + are present (MergeFile: collect parts until the full file is ready) + +This pipeline is a reference for the Step API shape (docs/design.md, +section 8) and the blob-store integration pattern, not a performance +example. +""" + +from __future__ import annotations + +from revenant.config import StageConfig +from revenant.step import Step + + +class SplitFile(Step): + """Stateless: split one input file into blob-backed parts.""" + + def process(self, payload, state): + input = payload.get("input", "") + with open(input) as f: + lines = [l for l in f] + total = len(lines) + for l in lines: + # Each line becomes a separate blob-backed part that can be + # merged later once the full input has arrived. + part = self.blob_store.write(bytes(l, 'utf8')) + yield {'input': input, 'part': part, 'total': total} + return state + +class MergeFile(Step): + """Stateful: collect blob parts until the file can be reassembled.""" + + def process(self, payload, state): + input = payload.get("input") + part = payload.get("part") + total = payload.get("total") + state = state or {"parts": {}} + parts = state['parts'].setdefault(input, []) + parts.append(part) + state["parts"][input] = parts + if len(parts) < total: + # Keep gathering fragments for this input until all expected + # parts have arrived. + return state + # Once every fragment is present, read them back from blob storage + # and emit the merged blob for the original input. + texts = [self.blob_store.resolve(p).read_text() for p in parts] + text = "".join(texts) + result = self.blob_store.write(bytes(text, 'utf8')) + yield {"input": input, "total":total, "result":result} + del(state["parts"][input]) + return state + + +PIPELINE = [ + StageConfig(name="split", step_class=SplitFile, upstream="input"), + StageConfig(name="merge", step_class=MergeFile, upstream="split"), +] diff --git a/pyproject.toml b/pyproject.toml index e61df78..43ee227 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,3 +28,4 @@ where = ["src"] [tool.pytest.ini_options] testpaths = ["tests"] +pythonpath = ["."] diff --git a/src/revenant/blob_store.py b/src/revenant/blob_store.py new file mode 100644 index 0000000..c0af6a6 --- /dev/null +++ b/src/revenant/blob_store.py @@ -0,0 +1,88 @@ +""" +Content-addressed blob storage for stage outputs that shouldn't be +embedded directly in .jsonl payloads (e.g. images, media files). See +docs/design-blob-storage.md for the full rationale. + +One BlobStore per stage process, constructed once at stage-runner +startup and attached to the Step instance as `step.blob_store`. +""" + +from __future__ import annotations + +import hashlib +import os +import uuid +from pathlib import Path + + +class BlobStore: + def __init__(self, blobs_dir: Path, state_dir: Path): + self.blobs_dir = blobs_dir + self.state_dir = state_dir + self.scratch_dir = blobs_dir / ".scratch" + self.blobs_dir.mkdir(parents=True, exist_ok=True) + self.scratch_dir.mkdir(parents=True, exist_ok=True) + + def reserve_path(self, suffix: str = "") -> Path: + """Return a scratch path an external tool can write to directly.""" + return self.scratch_dir / f"{uuid.uuid4().hex}{suffix}" + + def reserve_dir(self) -> Path: + """Same idea, for tools that produce many output files at once + (e.g. ffmpeg segmenting one input into N chunk files).""" + d = self.scratch_dir / uuid.uuid4().hex + d.mkdir(parents=True, exist_ok=True) + return d + + def write(self, data: bytes, suffix: str = "") -> str: + """Convenience wrapper: write in-memory bytes and commit in one call.""" + scratch_path = self.reserve_path(suffix=suffix) + fd = os.open(scratch_path, os.O_WRONLY | os.O_CREAT, 0o644) + try: + os.write(fd, data) + os.fsync(fd) + finally: + os.close(fd) + return self.commit(scratch_path, suffix=suffix) + + def commit(self, scratch_path: Path, suffix: str | None = None) -> str: + """Hash a completed scratch file and move it into content-addressed + storage. Returns a path relative to state_dir, suitable for storing + directly in a payload dict. + + `scratch_path` must already be fully written and fsync'd before + calling this -- commit() does not fsync the source file itself, + only the containing-directory fsync implied by os.replace on most + Linux filesystems. Callers writing via write() get this for free; + callers using reserve_path()/reserve_dir() directly (e.g. an + external subprocess) are responsible for the tool having finished + and flushed before commit() is called. + """ + digest = _sha256_file(scratch_path) + actual_suffix = suffix if suffix is not None else scratch_path.suffix + shard = digest[:2] + name = f"{digest}{actual_suffix}" + target_dir = self.blobs_dir / shard + target_dir.mkdir(parents=True, exist_ok=True) + final_path = target_dir / name + + if final_path.exists(): + # Identical content already stored (e.g. a deterministic retry + # after a crash) -- discard the duplicate scratch file. + scratch_path.unlink(missing_ok=True) + else: + os.replace(scratch_path, final_path) # atomic: same filesystem + + return str(final_path.relative_to(self.state_dir)) + + def resolve(self, relative_path: str) -> Path: + """Turn a payload's stored relative path back into a real path.""" + return self.state_dir / relative_path + + +def _sha256_file(path: Path) -> str: + h = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(1024 * 1024), b""): + h.update(chunk) + return h.hexdigest() diff --git a/src/revenant/config.py b/src/revenant/config.py index 4b2eaed..50e95db 100644 --- a/src/revenant/config.py +++ b/src/revenant/config.py @@ -69,3 +69,6 @@ def upstream_output_path(self, state_dir: Path) -> Path: if self.upstream == "input": return state_dir / "input.jsonl" return state_dir / f"{self.upstream}.jsonl" + + def blobs_dir(self, state_dir: Path) -> Path: + return state_dir / f"{self.name}.blobs" diff --git a/src/revenant/stage_runner.py b/src/revenant/stage_runner.py index 934b516..aa46146 100644 --- a/src/revenant/stage_runner.py +++ b/src/revenant/stage_runner.py @@ -27,6 +27,7 @@ read_last_checkpoint_line, ) from revenant.step import RetryableError, SkipItem +from revenant.blob_store import BlobStore POLL_INTERVAL_SECONDS = 1.0 @@ -137,6 +138,8 @@ def run_stage( last_consumed_seq, last_emitted_seq, saved_state = load_resume_point(stage, state_dir) step = stage.step_class() + step.blob_store = BlobStore(stage.blobs_dir(state_dir), state_dir) + step.state_dir = state_dir state = step.load(saved_state) input_final_seq = read_input_final_seq(state_dir) diff --git a/src/revenant/step.py b/src/revenant/step.py index cd8f4f8..3db8225 100644 --- a/src/revenant/step.py +++ b/src/revenant/step.py @@ -35,6 +35,25 @@ class Step: untouched, and the stage process exits non-zero. On restart, the same item is retried from scratch. This is intentional — see docs/design.md section 8. + + Attributes set by stage runner: + blob_store: BlobStore -- for writing/reading binary artifacts + too large or unsuitable to embed in + .jsonl payloads. See + docs/design-blob-storage.md. + state_dir: Path -- the pipeline's state directory, for + resolving blob paths read from an + upstream payload. + + IMPORTANT: a payload field containing a blob path (as returned by + self.blob_store.write()/commit()) must not be forwarded unchanged into + this step's own yield. If downstream needs the referenced content to + persist past this stage, re-write it via self.blob_store into this + stage's own blob directory. Blob paths are one-hop only -- see + docs/design-blob-storage.md section 1. This is not enforced by the + framework; violating it will not fail loudly today, but will make a + blob unsafe to reclaim whenever garbage collection is eventually + implemented. """ def load(self, saved_state: Any | None) -> Any: diff --git a/tests/examples/test_blob_split_merge.py b/tests/examples/test_blob_split_merge.py new file mode 100755 index 0000000..2507d8e --- /dev/null +++ b/tests/examples/test_blob_split_merge.py @@ -0,0 +1,76 @@ +import json + +from revenant.config import StageConfig +from revenant.io_utils import make_checkpoint_line, make_record_line +from revenant.stage_runner import run_stage + +from examples.blob_split_merge.pipeline import SplitFile, MergeFile + + +def _write_input_jsonl(path, payloads): + path.parent.mkdir(parents=True, exist_ok=True) + lines = [ + json.dumps(make_record_line(seq=i, src_seq=i, parent_seq=i, payload=p)) + for i, p in enumerate(payloads, start=1) + ] + lines.append(json.dumps(make_checkpoint_line(last_consumed_seq=len(payloads), last_emitted_seq=len(payloads)))) + path.write_text("\n".join(lines) + "\n") + + +def _run_split_then_merge(tmp_path, source_file): + state_dir = tmp_path / "state" + state_dir.mkdir() + + split_stage = StageConfig(name="split", step_class=SplitFile, upstream="input") + merge_stage = StageConfig(name="merge", step_class=MergeFile, upstream="split") + pipeline = [split_stage, merge_stage] + + _write_input_jsonl(split_stage.upstream_output_path(state_dir), [{"input": str(source_file)}]) + + run_stage(split_stage, state_dir, pipeline=pipeline) + run_stage(merge_stage, state_dir, pipeline=pipeline) + + output_path = merge_stage.output_path(state_dir) + if not output_path.exists(): + return state_dir, [] + + output_lines = [ + json.loads(line) + for line in output_path.read_text().splitlines() + if line.strip() + ] + return state_dir, [line for line in output_lines if line.get("type") == "record"] + + +def test_empty_file_produces_no_merge_output(tmp_path): + source_file = tmp_path / "empty.txt" + source_file.write_text("") + + state_dir, records = _run_split_then_merge(tmp_path, source_file) + + # SplitFile has no lines to yield for an empty file, so nothing ever + # reaches MergeFile for this input -- zero parts is not the same as + # "one part with total=0", and no output record should be produced. + assert records == [] + + +def test_single_line_file_merges_to_one_result(tmp_path): + source_file = tmp_path / "single_line.txt" + source_file.write_text("only one line\n") + + state_dir, records = _run_split_then_merge(tmp_path, source_file) + + assert len(records) == 1 + resolved = state_dir / records[0]["payload"]["result"] + assert resolved.read_text() == "only one line\n" + + +def test_multi_line_file_merges_all_parts_in_order(tmp_path): + source_file = tmp_path / "multi_line.txt" + source_file.write_text("line one\nline two\nline three\n") + + state_dir, records = _run_split_then_merge(tmp_path, source_file) + + assert len(records) == 1 + resolved = state_dir / records[0]["payload"]["result"] + assert resolved.read_text() == "line one\nline two\nline three\n" diff --git a/tests/test_blob_store.py b/tests/test_blob_store.py new file mode 100755 index 0000000..52f1643 --- /dev/null +++ b/tests/test_blob_store.py @@ -0,0 +1,106 @@ +import json + +from revenant.blob_store import BlobStore +from revenant.config import StageConfig +from revenant.io_utils import make_checkpoint_line, make_record_line +from revenant.stage_runner import run_stage +from revenant.step import Step + + +def test_write_round_trips_bytes(tmp_path): + state_dir = tmp_path / "state" + state_dir.mkdir() + store = BlobStore(state_dir / "A.blobs", state_dir) + + relative_path = store.write(b"hello world", suffix=".txt") + + resolved = store.resolve(relative_path) + assert resolved.exists() + assert resolved.read_bytes() == b"hello world" + # Path returned must be relative to state_dir, as specified. + assert resolved == (state_dir / relative_path) + assert not relative_path.startswith("/") + assert resolved.is_absolute() + + +def test_identical_writes_dedup_to_same_path(tmp_path): + state_dir = tmp_path / "state" + state_dir.mkdir() + store = BlobStore(state_dir / "A.blobs", state_dir) + + first_path = store.write(b"deterministic payload", suffix=".bin") + second_path = store.write(b"deterministic payload", suffix=".bin") + + assert first_path == second_path + resolved = store.resolve(first_path) + assert resolved.exists() + assert resolved.read_bytes() == b"deterministic payload" + + # Only one file should exist under the content-addressed shard -- + # the second write must have discarded its scratch copy, not created + # a duplicate blob. + shard_dir = resolved.parent + assert len(list(shard_dir.iterdir())) == 1 + + +def test_reserve_dir_and_commit_multi_output(tmp_path): + state_dir = tmp_path / "state" + state_dir.mkdir() + store = BlobStore(state_dir / "A.blobs", state_dir) + + out_dir = store.reserve_dir() + assert out_dir.exists() + assert out_dir.is_dir() + + file_a = out_dir / "chunk_0001.ts" + file_b = out_dir / "chunk_0002.ts" + file_a.write_bytes(b"chunk one") + file_b.write_bytes(b"chunk two") + + path_a = store.commit(file_a) + path_b = store.commit(file_b) + + assert path_a != path_b + assert store.resolve(path_a).read_bytes() == b"chunk one" + assert store.resolve(path_b).read_bytes() == b"chunk two" + # Scratch files should have been moved out, not copied. + assert not file_a.exists() + assert not file_b.exists() + + +def test_run_stage_persists_blob_path_in_committed_record(tmp_path): + class WriteBlobStep(Step): + def process(self, payload, state): + path = self.blob_store.write(bytes(payload["text"], "utf8"), suffix=".txt") + yield {"blob_path": path} + return state + + state_dir = tmp_path / "state" + state_dir.mkdir() + stage = StageConfig(name="A", step_class=WriteBlobStep, upstream="input") + + input_path = stage.upstream_output_path(state_dir) + input_path.write_text( + "\n".join( + [ + json.dumps(make_record_line(seq=1, src_seq=1, parent_seq=1, payload={"text": "hello"})), + json.dumps(make_checkpoint_line(last_consumed_seq=1, last_emitted_seq=1, state=None)), + ] + ) + + "\n" + ) + + run_stage(stage, state_dir) + + output_lines = [ + json.loads(line) for line in stage.output_path(state_dir).read_text().splitlines() if line.strip() + ] + record_lines = [line for line in output_lines if line.get("type") == "record"] + assert len(record_lines) == 1 + + blob_path = record_lines[0]["payload"]["blob_path"] + assert blob_path.startswith("A.blobs/") + + resolved = state_dir / blob_path + assert resolved.exists() + assert resolved.read_text() == "hello"