From 1bed241c4185d16d0fec5d980fb65f29cc7ae8b6 Mon Sep 17 00:00:00 2001 From: "Flemming N. Larsen" Date: Mon, 3 Aug 2026 23:08:50 +0200 Subject: [PATCH] fix durable result receipts --- .github/workflows/ingest.yml | 32 +++++++++++------ CONTRIBUTING.md | 2 +- README.md | 2 +- scripts/ingest.py | 70 ++++++++++++++++++++++++++++-------- tests/test_rumble_data.py | 35 ++++++++++++++++-- 5 files changed, 112 insertions(+), 29 deletions(-) diff --git a/.github/workflows/ingest.yml b/.github/workflows/ingest.yml index e5775d8..4eab50d 100644 --- a/.github/workflows/ingest.yml +++ b/.github/workflows/ingest.yml @@ -24,7 +24,7 @@ jobs: - uses: actions/setup-python@v5 with: python-version: '3.12' - - name: Drain labelled issue inbox + - name: Stage labelled issue inbox env: GH_TOKEN: ${{ github.token }} run: | @@ -36,26 +36,36 @@ jobs: number=$(jq -r .number .inbox/issue.json) account=$(jq -r .author.login .inbox/issue.json) jq -r .body .inbox/issue.json > .inbox/body.md - if python scripts/extract_envelope.py --body .inbox/body.md --output .inbox/envelope.json && python scripts/ingest.py --root . --account "$account" --input .inbox/envelope.json > .inbox/outcome.txt 2>&1; then - printf 'Processed result batch:\n\n```text\n' > .inbox/comment.md - cat .inbox/outcome.txt >> .inbox/comment.md - printf '\n```\n' >> .inbox/comment.md + : > .inbox/outcome.txt + if python scripts/extract_envelope.py --body .inbox/body.md --output .inbox/envelope.json >> .inbox/outcome.txt 2>&1 && python scripts/ingest.py --root . --account "$account" --input .inbox/envelope.json >> .inbox/outcome.txt 2>&1; then + printf 'Processed result batch:\n\n```text\n' > ".inbox/comment-$number.md" + cat .inbox/outcome.txt >> ".inbox/comment-$number.md" + printf '\n```\n' >> ".inbox/comment-$number.md" else - printf 'Rejected result batch before individual records could be read:\n\n```text\n' > .inbox/comment.md - cat .inbox/outcome.txt >> .inbox/comment.md - printf '\n```\n' >> .inbox/comment.md + printf 'Rejected result batch before individual records could be read:\n\n```text\n' > ".inbox/comment-$number.md" + cat .inbox/outcome.txt >> ".inbox/comment-$number.md" + printf '\n```\n' >> ".inbox/comment-$number.md" fi - gh issue comment "$number" --repo "$GITHUB_REPOSITORY" --body-file .inbox/comment.md - gh issue close "$number" --repo "$GITHUB_REPOSITORY" + printf '%s\n' "$number" >> .inbox/processed-issues done < .inbox/issues - name: Regenerate projections and commit accepted facts run: | python scripts/aggregate.py --root . git config user.name 'github-actions[bot]' git config user.email '41898282+github-actions[bot]@users.noreply.github.com' - rm -rf .inbox for path in results leaderboard matchmaking clients.json site/data; do [ -e "$path" ] && git add -- "$path" done git diff --cached --quiet || git commit -m 'chore: ingest Rumble result batch' git push + - name: Publish durable receipts + env: + GH_TOKEN: ${{ github.token }} + run: | + if [ -f .inbox/processed-issues ]; then + while read -r number; do + gh issue comment "$number" --repo "$GITHUB_REPOSITORY" --body-file ".inbox/comment-$number.md" + gh issue close "$number" --repo "$GITHUB_REPOSITORY" + done < .inbox/processed-issues + fi + rm -rf .inbox diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 4ff8bc3..629f6ff 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -10,7 +10,7 @@ Create an issue titled `[result] `, apply the `result Each participant supplies cataloged `name` and `version`, `isTeam`, a 1224-system `rank`, `totalScore`, `survival`, `lastSurvivorBonus`, `bulletDamage`, `bulletKillBonus`, `ramDamage`, `ramKillBonus`, `firstPlaces`, `secondPlaces`, and `thirdPlaces`. Scores and place counts are non-negative signed 32-bit integers. `isTeam`, result-entry count, rank placement, and place-count totals must match the selected ranked game type. -The submitting issue account and `clientId` must be registered. Every bot must be active in `catalog.json`; the hourly synchronization workflow copies the reviewed Rumble bot catalog declared by that file. The drain workflow validates each record independently, keeps valid records when neighboring records are rejected, and closes every processed issue with a receipt line for every record. +The submitting issue account and `clientId` must be registered. Every bot must be active in `catalog.json`; the hourly synchronization workflow copies the reviewed Rumble bot catalog declared by that file. The drain workflow validates each record independently, keeps valid records when neighboring records are rejected, pushes accepted facts, and then closes every processed issue with a receipt line for every record. Retrying an identical retained result returns the same successful outcome without creating another fact; reusing a battle ID for different content is rejected. ## Rules diff --git a/README.md b/README.md index 857f209..a715584 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ Run the focused suite with `python -m unittest discover -s tests -v`. It uses on Before a client can submit ranked results, its forge account must be registered through a reviewed pull request adding `clients/.json`. The client then creates an issue labelled `result-submission`, with a `[result]` title and exactly one fenced JSON batch envelope. See [CONTRIBUTING.md](CONTRIBUTING.md) for the contract and limits. -Issue bodies are transport receipts, never durable storage. The only authoritative accepted result is a content-addressed JSON fact under `results/raw/`; projections are disposable and reproducible. +Issue bodies are transport, never durable storage. The only authoritative accepted result is a content-addressed JSON fact under `results/raw/`; projections are disposable and reproducible. Successful receipt comments are published only after accepted facts are pushed, and identical retries are acknowledged idempotently. ## Forking and operations diff --git a/scripts/ingest.py b/scripts/ingest.py index bb601b8..d1fa018 100644 --- a/scripts/ingest.py +++ b/scripts/ingest.py @@ -6,8 +6,8 @@ import argparse from pathlib import Path -from common import read_json, write_json -from validate import AcceptedResult, ValidationError, ValidationOutcome, validate_envelope +from common import content_hash, read_json, write_json +from validate import SCHEMA_VERSION, AcceptedResult, ValidationError, ValidationOutcome, validate_envelope def raw_path(root: Path, completed_at: str, digest: str) -> Path: @@ -18,16 +18,53 @@ def raw_path(root: Path, completed_at: str, digest: str) -> Path: return root / "results" / "raw" / date[0] / date[1] / f"{digest}.json" -def retained_values(root: Path, field: str) -> set[str]: - """Return one field's values from every retained raw fact and rollup record.""" - result: set[str] = set() +def retained_results(root: Path) -> dict[str, dict]: + """Return retained facts indexed by battle ID.""" + result: dict[str, dict] = {} for path in (root / "results" / "raw").rglob("*.json") if (root / "results" / "raw").exists() else []: - result.add(str(read_json(path).get(field))) + record = read_json(path) + result[str(record.get("battleId"))] = record for path in (root / "results" / "rollups").rglob("*.json") if (root / "results" / "rollups").exists() else []: - result.update(str(record.get(field)) for record in read_json(path).get("results", [])) + for record in read_json(path).get("results", []): + result[str(record.get("battleId"))] = record return result +def validate_with_idempotent_retries(root: Path, envelope: object, *, account: str, + retained: dict[str, dict]) -> list[ValidationOutcome]: + """Validate new records while recovering successes for identical retained facts.""" + if not isinstance(envelope, dict) or not isinstance(envelope.get("results"), list): + return validate_envelope(root, envelope, account=account) + recovered: dict[int, ValidationOutcome] = {} + pending_records: list[object] = [] + pending_indexes: list[int] = [] + recoverable_envelope = envelope.get("schemaVersion") == SCHEMA_VERSION and 0 < len(envelope["results"]) <= 60 + for index, record in enumerate(envelope["results"]): + battle_id = str(record.get("battleId")) if isinstance(record, dict) and record.get("battleId") is not None else None + existing = retained.get(battle_id) if battle_id is not None else None + client = record.get("client") if isinstance(record, dict) else None + identical = ( + recoverable_envelope + and existing is not None + and existing.get("submittedBy") == account + and content_hash(record) == existing.get("payloadHash") + and isinstance(client, dict) + and envelope.get("clientId") == client.get("id") + and envelope.get("clientVersion") == client.get("version") + ) + if identical: + accepted = AcceptedResult(existing, content_hash(existing)) + recovered[index] = ValidationOutcome(index=index, battle_id=battle_id, accepted=accepted) + else: + pending_indexes.append(index) + pending_records.append(record) + if pending_records: + pending_envelope = envelope | {"results": pending_records} + for index, outcome in zip(pending_indexes, validate_envelope(root, pending_envelope, account=account)): + recovered[index] = ValidationOutcome(index=index, battle_id=outcome.battle_id, accepted=outcome.accepted, error=outcome.error) + return [recovered[index] for index in range(len(envelope["results"]))] + + def format_outcome(outcome: ValidationOutcome) -> str: """Render one stable receipt line for a submitted record.""" label = outcome.battle_id or f"result[{outcome.index}]" @@ -36,9 +73,9 @@ def format_outcome(outcome: ValidationOutcome) -> str: def ingest(root: Path, envelope: object, *, account: str) -> list[ValidationOutcome]: """Append every independently accepted record and return all receipt outcomes.""" - outcomes = validate_envelope(root, envelope, account=account) - battle_ids = retained_values(root, "battleId") - payload_hashes = retained_values(root, "payloadHash") + retained = retained_results(root) + outcomes = validate_with_idempotent_retries(root, envelope, account=account, retained=retained) + payload_hashes = {str(record.get("payloadHash")) for record in retained.values()} persisted: list[ValidationOutcome] = [] for outcome in outcomes: if outcome.accepted is None: @@ -46,10 +83,15 @@ def ingest(root: Path, envelope: object, *, account: str) -> list[ValidationOutc continue item: AcceptedResult = outcome.accepted battle_id = str(item.record["battleId"]) - if battle_id in battle_ids: + payload_hash = str(item.record["payloadHash"]) + existing = retained.get(battle_id) + if existing is item.record: + persisted.append(outcome) + continue + if battle_id in retained: persisted.append(ValidationOutcome(index=outcome.index, battle_id=battle_id, error="duplicate battleId already retained")) continue - if item.record["payloadHash"] in payload_hashes: + if payload_hash in payload_hashes: persisted.append(ValidationOutcome(index=outcome.index, battle_id=battle_id, error="duplicate payload hash already retained")) continue path = raw_path(root, str(item.record["completedAt"]), item.digest) @@ -57,8 +99,8 @@ def ingest(root: Path, envelope: object, *, account: str) -> list[ValidationOutc persisted.append(ValidationOutcome(index=outcome.index, battle_id=battle_id, error="duplicate payload hash already retained")) continue write_json(path, item.record) - battle_ids.add(battle_id) - payload_hashes.add(str(item.record["payloadHash"])) + retained[battle_id] = item.record + payload_hashes.add(payload_hash) persisted.append(outcome) return persisted diff --git a/tests/test_rumble_data.py b/tests/test_rumble_data.py index ded9d97..b29f618 100644 --- a/tests/test_rumble_data.py +++ b/tests/test_rumble_data.py @@ -75,17 +75,48 @@ def testRDA001_IntegrationPositive_valid_records_survive_invalid_batch_neighbors self.assertIn("rank", outcomes[1].error) self.assertEqual(1, len(list((self.root / "results/raw").rglob("*.json")))) - def testRDA002_IntegrationNegative_structural_and_duplicate_records_never_persist(self) -> None: + def testRDA002_IntegrationNegative_structural_records_never_persist(self) -> None: malformed = self.envelope()["results"][0] malformed.pop("rounds") outcome = ingest(self.root, {**self.envelope(), "results": [malformed]}, account="alice")[0] self.assertIsNone(outcome.accepted) self.assertIn("rounds", outcome.error) self.assertFalse((self.root / "results/raw").exists()) + + def testRDA001_IntegrationPositive_identical_retry_is_idempotently_accepted(self) -> None: + self.assertTrue(ingest(self.root, self.envelope(), account="alice")[0].accepted) + retry = ingest(self.root, self.envelope(), account="alice")[0] + self.assertTrue(retry.accepted) + self.assertEqual(1, len(list((self.root / "results/raw").rglob("*.json")))) + + def testRDA001_IntegrationPositive_identical_retry_survives_current_pin_change(self) -> None: self.assertTrue(ingest(self.root, self.envelope(), account="alice")[0].accepted) - duplicate = ingest(self.root, self.envelope(), account="alice")[0] + self.write("engine.json", {"schemaVersion": 1, "behaviorVersion": 2, "gameTypes": {"1v1": {"rounds": 35, "battlefield": [800, 600], "participants": 2}}}) + retry = ingest(self.root, self.envelope(), account="alice")[0] + self.assertTrue(retry.accepted) + self.assertEqual(1, len(list((self.root / "results/raw").rglob("*.json")))) + + def testRDA002_IntegrationNegative_conflicting_battle_id_is_rejected(self) -> None: + self.assertTrue(ingest(self.root, self.envelope(), account="alice")[0].accepted) + conflicting = self.envelope() + conflicting["results"][0]["completedAt"] = "2026-08-02T12:01:00Z" + duplicate = ingest(self.root, conflicting, account="alice")[0] self.assertIsNone(duplicate.accepted) self.assertIn("duplicate battleId", duplicate.error) + self.assertEqual(1, len(list((self.root / "results/raw").rglob("*.json")))) + + def testRDA002_IntegrationNegative_duplicate_within_one_batch_is_rejected(self) -> None: + envelope = self.envelope() + envelope["results"].append(dict(envelope["results"][0])) + outcomes = ingest(self.root, envelope, account="alice") + self.assertTrue(outcomes[0].accepted) + self.assertIsNone(outcomes[1].accepted) + self.assertIn("duplicate battleId", outcomes[1].error) + self.assertEqual(1, len(list((self.root / "results/raw").rglob("*.json")))) + + def testArch_successful_receipts_follow_fact_publication(self) -> None: + workflow = (ROOT / ".github/workflows/ingest.yml").read_text(encoding="utf-8") + self.assertLess(workflow.index("git push"), workflow.index("gh issue comment")) def testRDA002_IntegrationNegative_rejects_each_documented_structural_violation(self) -> None: invalid_records = []