diff --git a/solstone/apps/observer/routes.py b/solstone/apps/observer/routes.py index 4ddf91b87..40bc3fe1a 100644 --- a/solstone/apps/observer/routes.py +++ b/solstone/apps/observer/routes.py @@ -878,33 +878,6 @@ def _process_ingest_files( INGEST_SIDECAR_CONFLICT.status, ) - if plan.status == "duplicate": - save_ingest_plan(plan, allow_reentry=False) - append_history_record(key_prefix, day, _sync_record_for_plan(plan)) - logger.info( - "Observer ingest outcome=candidate_matched day=%s stream=%s requested=%s segment=%s", - day, - stream, - segment, - plan.segment, - ) - - observer["last_seen"] = now_ms() - observer["stats"]["duplicates_rejected"] = ( - observer["stats"].get("duplicates_rejected", 0) + 1 - ) - clear_ingest_rejection(observer) - save_observer(observer) - - return ( - { - "status": "duplicate", - "existing_segment": plan.existing_segment or plan.segment, - "message": "All files already received", - }, - 200, - ) - if plan.status == "storage_failed": logger.error( "No available segment slot for %s/%s/%s from %s after %s attempts", @@ -989,6 +962,39 @@ def _process_ingest_files( second_result = save_ingest_plan(plan, allow_reentry=False) apply_result = _merge_apply_results(first_result, second_result) + if plan.status == "duplicate": + if not history_recorded: + append_history_record( + key_prefix, + day, + _sync_record_for_plan( + plan, records=_records_with_apply_result(plan.records, apply_result) + ), + ) + logger.info( + "Observer ingest outcome=candidate_matched day=%s stream=%s requested=%s segment=%s", + day, + stream, + segment, + plan.segment, + ) + + observer["last_seen"] = now_ms() + observer["stats"]["duplicates_rejected"] = ( + observer["stats"].get("duplicates_rejected", 0) + 1 + ) + clear_ingest_rejection(observer) + save_observer(observer) + + return ( + { + "status": "duplicate", + "existing_segment": plan.existing_segment or plan.segment, + "message": "All files already received", + }, + 200, + ) + if not history_recorded: append_history_record( key_prefix, @@ -1425,7 +1431,7 @@ def ingest_segments(day: str) -> Any: if segment not in segments: segments[segment] = { "key": segment, - "files_by_sha": {}, # Keyed by sha256 for deduplication + "files_by_identity": {}, # Keyed by (name, sha256) for deduplication } if segment_original: segments[segment]["original_key"] = segment_original @@ -1456,19 +1462,20 @@ def ingest_segments(day: str) -> Any: day_dir, stream, segment, written, size ) - # Deduplicate by sha256 - later uploads overwrite earlier - segments[segment]["files_by_sha"][sha256] = file_info + # Deduplicate exact same name+sha records; identical bytes under + # different names must remain separately corroboratable. + segments[segment]["files_by_identity"][(written, sha256)] = file_info - # Convert files_by_sha dicts to lists and sort by segment key + # Convert files_by_identity dicts to lists and sort by segment key result = [] for segment_data in sorted(segments.values(), key=lambda s: s["key"]): - if not segment_data["files_by_sha"]: + if not segment_data["files_by_identity"]: continue segment_key = segment_data["key"] entry = { "key": segment_key, "observed": segment_key in observed_segments, - "files": list(segment_data["files_by_sha"].values()), + "files": list(segment_data["files_by_identity"].values()), } if "original_key" in segment_data: entry["original_key"] = segment_data["original_key"] diff --git a/solstone/apps/observer/tests/test_routes.py b/solstone/apps/observer/tests/test_routes.py index 9247a0a39..365d50986 100644 --- a/solstone/apps/observer/tests/test_routes.py +++ b/solstone/apps/observer/tests/test_routes.py @@ -10,6 +10,8 @@ import io import json import os import shutil +import threading +import time from pathlib import Path from unittest.mock import MagicMock @@ -3468,6 +3470,43 @@ def test_ingest_manifest_absence_or_corruption_never_duplicates_from_cache( assert corrupt.get_json()["status"] == "ok" assert (segment_dir / "audio.flac").read_bytes() == audio + variants = [ + ("unknown-version", {"schema_version": 999}), + ("missing-version", {}), + ("string-version", {"schema_version": "1"}), + ("bad-file-entry", {"schema_version": 1, "entry": "not-a-dict"}), + ] + for suffix, manifest_shape in variants: + variant_audio = f"manifest {suffix} audio".encode() + variant_key = _create_observer(env, f"manifest-{suffix}-test") + first = _upload_audio(env, variant_key, variant_audio) + assert first.status_code == 200 + variant_dir = _day_dir(env) / f"manifest-{suffix}-test" / "120000_300" + _write_audio_processing_sidecar(variant_dir, input_size=len(variant_audio)) + (variant_dir / "audio.flac").unlink() + manifest_files = { + "audio.flac": { + "sha256": hashlib.sha256(variant_audio).hexdigest(), + "size": len(variant_audio), + } + } + manifest = { + "requested_segment": "120000_300", + "files": manifest_files, + **manifest_shape, + } + if suffix == "bad-file-entry": + manifest["files"] = {"audio.flac": manifest_shape["entry"]} + (variant_dir / "ingest.json").write_text( + json.dumps(manifest) + "\n", encoding="utf-8" + ) + + retry = _upload_audio(env, variant_key, variant_audio) + body = retry.get_json() + assert retry.status_code == 200 + assert body["status"] == "ok" + assert (variant_dir / "audio.flac").read_bytes() == variant_audio + def test_ingest_manifest_different_hash_with_terminal_proof_disqualifies_candidate( observer_env, @@ -3659,59 +3698,104 @@ def test_ingest_manifest_accumulates_and_duplicates_after_media_deleted( def test_ingest_held_content_never_mints_for_request_shape(observer_env): - """AC-9: held content resolves in place for duplicate/additive/conflict shapes.""" + """AC-9: held content resolves in place across request shapes.""" env = observer_env() - key = _create_observer(env, "held-no-mint-test") - audio = b"held content" - - first = _upload_audio(env, key, audio) - assert first.status_code == 200 - stream_dir = _day_dir(env) / "held-no-mint-test" - - additive = env.client.post( - "/app/observer/ingest", - headers={"Authorization": f"Bearer {key}"}, - data={ - "day": "20250103", - "segment": "120000_300", - "files": [ - (io.BytesIO(audio), "audio.flac"), - (io.BytesIO(b"notes"), "notes.txt"), + audio = b"held media content" + tmux = b'{"line": 1}\n' + cases = [ + ( + "media-only-exact", + [(audio, "audio.flac")], + [(audio, "audio.flac")], + "120000_300", + "duplicate", + ), + ( + "media-new-sidecar", + [(audio, "audio.flac")], + [(audio, "audio.flac"), (b"notes", "notes.txt")], + "120000_300", + "ok", + ), + ( + "media-reserved", + [(audio, "audio.flac")], + [ + (audio, "audio.flac"), + ( + b'{"stream":"held-shape-media-reserved","prev_day":null,' + b'"prev_segment":null,"seq":1}\n', + "stream.json", + ), ], - }, - ) - assert additive.status_code == 200 - assert additive.get_json()["status"] == "ok" + "120000_300", + "duplicate", + ), + ( + "media-laddered-key", + [(audio, "audio.flac")], + [(audio, "audio.flac")], + "120000_301", + "duplicate", + ), + ( + "media-different-len", + [(audio, "audio.flac")], + [(audio, "audio.flac")], + "120000_999", + "duplicate", + ), + ( + "media-less-exact", + [(tmux, "tmux.jsonl")], + [(tmux, "tmux.jsonl")], + "120000_300", + "duplicate", + ), + ( + "media-less-laddered-key", + [(tmux, "tmux.jsonl")], + [(tmux, "tmux.jsonl")], + "120000_301", + "duplicate", + ), + ] - conflict = env.client.post( - "/app/observer/ingest", - headers={"Authorization": f"Bearer {key}"}, - data={ - "day": "20250103", - "segment": "120000_300", - "files": [ - (io.BytesIO(audio), "audio.flac"), - (io.BytesIO(b"changed notes"), "notes.txt"), - ], - }, - ) - assert conflict.status_code == 409 + for suffix, initial_files, retry_files, requested_segment, expected_status in cases: + observer_name = f"held-shape-{suffix}" + key = _create_observer(env, observer_name) + first = env.client.post( + "/app/observer/ingest", + headers={"Authorization": f"Bearer {key}"}, + data={ + "day": "20250103", + "segment": "120000_300", + "files": [ + (io.BytesIO(content), name) for content, name in initial_files + ], + }, + ) + assert first.status_code == 200 - reserved = env.client.post( - "/app/observer/ingest", - headers={"Authorization": f"Bearer {key}"}, - data={ - "day": "20250103", - "segment": "120000_300", - "files": [ - (io.BytesIO(audio), "audio.flac"), - (io.BytesIO(b'{"stream":"other","seq":9}\n'), "stream.json"), - ], - }, - ) - assert reserved.status_code == 200 - assert reserved.get_json()["status"] == "duplicate" - assert sorted(path.name for path in stream_dir.iterdir()) == ["120000_300"] + retry = env.client.post( + "/app/observer/ingest", + headers={"Authorization": f"Bearer {key}"}, + data={ + "day": "20250103", + "segment": requested_segment, + "files": [(io.BytesIO(content), name) for content, name in retry_files], + }, + ) + body = retry.get_json() + assert retry.status_code == 200 + assert body["status"] == expected_status + if expected_status == "duplicate": + assert body["existing_segment"] == "120000_300" + else: + assert body["segment"] == "120000_300" + assert sorted( + path.name for path in (_day_dir(env) / observer_name).iterdir() + ) == ["120000_300"] def test_ingest_create_collision_reenters_once(observer_env, monkeypatch): @@ -3776,6 +3860,153 @@ def test_ingest_create_collision_identical_bytes_stays_on_single_dir( assert sorted(path.name for path in stream_dir.iterdir()) == ["120000_300"] +def test_ingest_multifile_create_collision_reenters_once(observer_env, monkeypatch): + """AC-11: a second-file race re-enters once and preserves landed bytes.""" + env = observer_env() + key = _create_observer(env, "multifile-race-test") + audio = b"loser audio" + screen = b"loser screen" + winner_screen = b"winner screen" + real_write = observer_utils.write_bytes_exclusive + calls = {"count": 0} + + def race_on_second_file(path: Path, data: bytes, *, mode=None): + calls["count"] += 1 + if calls["count"] == 2: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(winner_screen) + raise FileExistsError(path) + return real_write(path, data, mode=mode) + + monkeypatch.setattr(observer_utils, "write_bytes_exclusive", race_on_second_file) + + resp = env.client.post( + "/app/observer/ingest", + headers={"Authorization": f"Bearer {key}"}, + data={ + "day": "20250103", + "segment": "120000_300", + "files": [ + (io.BytesIO(audio), "audio.flac"), + (io.BytesIO(screen), "screen.mp4"), + ], + }, + ) + body = resp.get_json() + + assert resp.status_code == 200 + assert body["status"] == "collision" + assert body["segment"] == "120000_301" + stream_dir = _day_dir(env) / "multifile-race-test" + assert (stream_dir / "120000_300" / "audio.flac").read_bytes() == audio + assert (stream_dir / "120000_300" / "screen.mp4").read_bytes() == winner_screen + assert (stream_dir / "120000_301" / "audio.flac").read_bytes() == audio + assert (stream_dir / "120000_301" / "screen.mp4").read_bytes() == screen + assert not (stream_dir / "120000_302").exists() + + +def test_ingest_duplicate_rechecks_held_files_before_response( + observer_env, monkeypatch +): + """Held files that vanish after planning heal instead of returning duplicate.""" + env = observer_env() + key = _create_observer(env, "held-recheck-test") + audio = b"held recheck audio" + + first = _upload_audio(env, key, audio) + assert first.status_code == 200 + segment_dir = _day_dir(env) / "held-recheck-test" / "120000_300" + real_recheck = observer_utils._held_files_are_current + calls = {"count": 0} + + def delete_once_then_check(path: Path, files): + calls["count"] += 1 + if calls["count"] == 1: + (path / files[0].written).unlink() + return real_recheck(path, files) + + monkeypatch.setattr( + observer_utils, "_held_files_are_current", delete_once_then_check + ) + + retry = _upload_audio(env, key, audio) + body = retry.get_json() + + assert retry.status_code == 200 + assert body["status"] == "ok" + assert body["segment"] == "120000_300" + assert (segment_dir / "audio.flac").read_bytes() == audio + + +def test_ingest_manifest_locked_merge_preserves_concurrent_additions( + observer_env, monkeypatch +): + """Concurrent manifest read-merge-write keeps both additive entries.""" + env = observer_env() + segment_dir = _day_dir(env) / "manifest-lock-test" / "120000_300" + segment_dir.mkdir(parents=True) + base = observer_utils.IngestFile( + submitted="audio.flac", + written="audio.flac", + content=b"base audio", + sha256=hashlib.sha256(b"base audio").hexdigest(), + ) + observer_utils.write_ingest_manifest( + segment_dir, requested_segment="120000_300", files=[base] + ) + + real_atomic_replace = observer_utils.atomic_replace + first_replace_entered = threading.Event() + release_first_replace = threading.Event() + calls = {"count": 0} + + def slow_first_replace(path: Path, data, *, mode=None): + if path.name == "ingest.json": + calls["count"] += 1 + if calls["count"] == 1: + first_replace_entered.set() + assert release_first_replace.wait(timeout=5) + return real_atomic_replace(path, data, mode=mode) + + monkeypatch.setattr(observer_utils, "atomic_replace", slow_first_replace) + + def ingest_file(name: str, content: bytes) -> observer_utils.IngestFile: + return observer_utils.IngestFile( + submitted=name, + written=name, + content=content, + sha256=hashlib.sha256(content).hexdigest(), + ) + + errors = [] + + def write_file(file: observer_utils.IngestFile) -> None: + try: + observer_utils.write_ingest_manifest( + segment_dir, requested_segment="120000_300", files=[file] + ) + except Exception as exc: # pragma: no cover - surfaced by assertion below + errors.append(exc) + + first_file = ingest_file("notes-a.txt", b"notes a") + second_file = ingest_file("notes-b.txt", b"notes b") + first_thread = threading.Thread(target=write_file, args=(first_file,)) + second_thread = threading.Thread(target=write_file, args=(second_file,)) + first_thread.start() + assert first_replace_entered.wait(timeout=5) + second_thread.start() + time.sleep(0.1) + release_first_replace.set() + first_thread.join(timeout=5) + second_thread.join(timeout=5) + + assert not first_thread.is_alive() + assert not second_thread.is_alive() + assert errors == [] + manifest = json.loads((segment_dir / "ingest.json").read_text(encoding="utf-8")) + assert set(manifest["files"]) == {"audio.flac", "notes-a.txt", "notes-b.txt"} + + def test_ingest_stream_chain_unchanged_for_existing_candidate_resolutions(observer_env): """AC-12: duplicate/additive/heal do not rewrite stream state or marker.""" env = observer_env() @@ -3848,6 +4079,49 @@ def test_ingest_duplicate_against_import_created_segment_is_corroborated( } +def test_ingest_duplicate_listing_keeps_same_sha_under_different_names(observer_env): + """AC-13: listing corroborates every submitted name, not only every sha.""" + env = observer_env() + key = _create_observer(env, "same-sha-names-test") + content = b"same bytes under two names" + sha256 = hashlib.sha256(content).hexdigest() + + first = env.client.post( + "/app/observer/ingest", + headers={"Authorization": f"Bearer {key}"}, + data={ + "day": "20250103", + "segment": "120000_300", + "files": [ + (io.BytesIO(content), "left.flac"), + (io.BytesIO(content), "right.flac"), + ], + }, + ) + assert first.status_code == 200 + + duplicate = env.client.post( + "/app/observer/ingest", + headers={"Authorization": f"Bearer {key}"}, + data={ + "day": "20250103", + "segment": "120000_300", + "files": [ + (io.BytesIO(content), "left.flac"), + (io.BytesIO(content), "right.flac"), + ], + }, + ) + assert duplicate.status_code == 200 + assert duplicate.get_json()["status"] == "duplicate" + + files = _listed_files_by_name(env, key) + assert files["left.flac"]["sha256"] == sha256 + assert files["left.flac"]["status"] == "present" + assert files["right.flac"]["sha256"] == sha256 + assert files["right.flac"]["status"] == "present" + + def test_ingest_returns_collision_status_when_adjusted(observer_env): """Test that collision resolution returns status='collision'.""" env = observer_env() diff --git a/solstone/apps/observer/utils.py b/solstone/apps/observer/utils.py index 8231d644c..b2ea040ab 100644 --- a/solstone/apps/observer/utils.py +++ b/solstone/apps/observer/utils.py @@ -29,7 +29,7 @@ from solstone.convey.reasons import ( from solstone.convey.utils import error_response from solstone.observe.protocol import OBSERVER_HANDLE_HEADER from solstone.think.contract.journal import ContractIssue -from solstone.think.journal_io import atomic_replace, write_bytes_exclusive +from solstone.think.journal_io import atomic_replace, hold_lock, write_bytes_exclusive from solstone.think.media import AUDIO_EXTENSIONS, VIDEO_EXTENSIONS from solstone.think.segment_files import ( INGEST_MANIFEST_NAME, @@ -745,6 +745,17 @@ def save_ingest_plan(plan: IngestPlan, *, allow_reentry: bool) -> IngestApplyRes """Apply a resolved ingest plan, writing client bytes create-exclusively.""" if plan.status == "duplicate": segment_dir = day_path(plan.day, create=False) / plan.stream / plan.segment + if plan.held_files and not _held_files_are_current( + segment_dir, plan.held_files + ): + if allow_reentry: + return IngestApplyResult( + files_written=[], + files_already_held=[], + bytes_written=0, + reenter_resolution=True, + ) + raise FileNotFoundError("held ingest files changed before response") if plan.held_files: write_ingest_manifest( segment_dir, @@ -790,6 +801,20 @@ def save_ingest_plan(plan: IngestPlan, *, allow_reentry: bool) -> IngestApplyRes files_written.append(file.written) bytes_written += file.size + # This closes the practical planner-to-response gap for held files, but it + # is not a full filesystem lock: a delete after this check can still race + # the history append/HTTP response. Closing that final gap needs a broader + # segment lock that observer ingest does not currently own. + if plan.held_files and not _held_files_are_current(segment_dir, plan.held_files): + if allow_reentry: + return IngestApplyResult( + files_written=files_written, + files_already_held=files_already_held, + bytes_written=bytes_written, + reenter_resolution=True, + ) + raise FileNotFoundError("held ingest files changed before response") + if plan.write_files or plan.held_files: write_ingest_manifest( segment_dir, @@ -808,18 +833,21 @@ def write_ingest_manifest( ) -> None: """Write the journal-authored ingest manifest for a segment.""" manifest_path = segment_dir / INGEST_MANIFEST_NAME - existing = _read_ingest_manifest(manifest_path) - manifest_files = dict(existing.get("files", {})) if existing else {} - for file in files: - if file.is_reserved: - continue - manifest_files[file.written] = {"sha256": file.sha256, "size": file.size} - manifest = { - "schema_version": INGEST_MANIFEST_SCHEMA_VERSION, - "requested_segment": requested_segment, - "files": manifest_files, - } - atomic_replace(manifest_path, json.dumps(manifest, indent=2, sort_keys=True) + "\n") + with hold_lock(manifest_path): + existing = _read_ingest_manifest(manifest_path) + manifest_files = dict(existing.get("files", {})) if existing else {} + for file in files: + if file.is_reserved: + continue + manifest_files[file.written] = {"sha256": file.sha256, "size": file.size} + manifest = { + "schema_version": INGEST_MANIFEST_SCHEMA_VERSION, + "requested_segment": requested_segment, + "files": manifest_files, + } + atomic_replace( + manifest_path, json.dumps(manifest, indent=2, sort_keys=True) + "\n" + ) def _content_identity_files(files: list[IngestFile]) -> list[IngestFile]: @@ -1041,9 +1069,18 @@ def _read_ingest_manifest(path: Path) -> dict[str, Any]: return {} if not isinstance(raw, dict): return {} + schema_version = raw.get("schema_version") + if ( + isinstance(schema_version, bool) + or not isinstance(schema_version, int) + or schema_version != INGEST_MANIFEST_SCHEMA_VERSION + ): + return {} files = raw.get("files") if not isinstance(files, dict): return {} + if not all(isinstance(entry, dict) for entry in files.values()): + return {} return raw @@ -1061,6 +1098,25 @@ def _manifest_entry_matches(entry: object, file: IngestFile) -> bool: ) +def _held_files_are_current(segment_dir: Path, files: list[IngestFile]) -> bool: + manifest_files = _manifest_files(segment_dir) + for file in files: + target = segment_dir / file.written + if target.exists(): + try: + if _file_sha256(target) == file.sha256: + continue + except OSError: + pass + manifest_entry = manifest_files.get(file.written) + if _manifest_entry_matches( + manifest_entry, file + ) and _has_terminal_processing_proof(target, file.size): + continue + return False + return True + + def _has_terminal_processing_proof(path: Path, size: int) -> bool: from solstone.apps.observer.processing_proof import has_terminal_processing_proof diff --git a/solstone/think/segment_files.py b/solstone/think/segment_files.py index b52dcce58..036ec1d75 100644 --- a/solstone/think/segment_files.py +++ b/solstone/think/segment_files.py @@ -5,9 +5,11 @@ STREAM_MARKER_NAME = "stream.json" INGEST_MANIFEST_NAME = "ingest.json" +INGEST_MANIFEST_LOCK_NAME = f"{INGEST_MANIFEST_NAME}.lock" RESERVED_SEGMENT_FILENAMES = frozenset( { STREAM_MARKER_NAME, INGEST_MANIFEST_NAME, + INGEST_MANIFEST_LOCK_NAME, } )