diff --git a/.github/workflows/publish-discord-backup.yml b/.github/workflows/publish-discord-backup.yml index b1d37cff..69c2b4f0 100644 --- a/.github/workflows/publish-discord-backup.yml +++ b/.github/workflows/publish-discord-backup.yml @@ -1,15 +1,21 @@ name: publish-discord-backup on: - schedule: - - cron: "*/15 * * * *" - - cron: "17 6 * * *" + # Admit a measured manual run before choosing a recurring cadence. workflow_dispatch: inputs: - reconcile_cloud: - description: Reconcile the configured private identity plan against Discord before publication - type: boolean - default: false + policy_commit: + description: Exact commit on the private maintainer-report-policy branch + type: string + required: true + policy_sha256: + description: SHA-256 of the reviewed private selector policy + type: string + required: true + since: + description: Optional UTC midnight less than 48 hours ago; defaults to previous UTC midnight + type: string + required: false permissions: contents: read @@ -37,11 +43,10 @@ jobs: go-version-file: go.mod cache: true - - name: Build native entry points + - name: Build native entry point timeout-minutes: 10 run: | go build -o "$RUNNER_TEMP/discrawl-native" ./cmd/discrawl - go build -o "$RUNNER_TEMP/discrawl-reconcile" scripts/reconcile_cloud_source.go - name: Configure runtime paths timeout-minutes: 1 @@ -52,6 +57,22 @@ jobs: echo "DB=$GITHUB_WORKSPACE/.discrawl-ci/discrawl.db" >> "$GITHUB_ENV" echo "BACKUP_REPO=$RUNNER_TEMP/discord-backup" >> "$GITHUB_ENV" echo "CHECKPOINT_RECEIPT=$RUNNER_TEMP/discrawl/checkpoint.json" >> "$GITHUB_ENV" + echo "ACQUISITION_POLICY=$RUNNER_TEMP/discrawl/report-selectors.json" >> "$GITHUB_ENV" + + - name: Verify pinned private acquisition policy + timeout-minutes: 2 + id: policy + env: + DISCORD_BACKUP_TOKEN: ${{ secrets.DISCORD_BACKUP_TOKEN }} + DISCRAWL_BACKUP_REPOSITORY: ${{ secrets.DISCRAWL_BACKUP_REPOSITORY }} + DISCRAWL_GUILD_ID: ${{ secrets.DISCRAWL_GUILD_ID }} + POLICY_COMMIT: ${{ inputs.policy_commit }} + POLICY_SHA256: ${{ inputs.policy_sha256 }} + HISTORY_SINCE: ${{ inputs.since }} + run: | + python3 scripts/private_discord_checkpoint.py policy --policy "$ACQUISITION_POLICY" \ + --policy-commit "$POLICY_COMMIT" --policy-sha256 "$POLICY_SHA256" \ + --since "$HISTORY_SINCE" >> "$GITHUB_OUTPUT" - name: Restore verified private checkpoint timeout-minutes: 10 @@ -66,42 +87,23 @@ jobs: python3 scripts/private_discord_checkpoint.py restore --prune --db "$DB" --receipt "$CHECKPOINT_RECEIPT" - name: Sync native Discord state - timeout-minutes: ${{ inputs.reconcile_cloud && 7 || 22 }} + timeout-minutes: 22 id: sync continue-on-error: true env: DISCORD_BOT_TOKEN: ${{ secrets.DISCORD_BOT_TOKEN }} DISCRAWL_GUILD_ID: ${{ secrets.DISCRAWL_GUILD_ID }} - RECONCILE_CLOUD: ${{ inputs.reconcile_cloud && '1' || '0' }} - MEMBER_REFRESH: ${{ github.event.schedule == '17 6 * * *' && '1' || '0' }} + POLICY_SHA256: ${{ inputs.policy_sha256 }} + HISTORY_SINCE: ${{ steps.policy.outputs.since }} run: | umask 077 test -n "$DISCORD_BOT_TOKEN" && test -n "$DISCRAWL_GUILD_ID" python3 scripts/private_discord_checkpoint.py run --timeout 60 -- \ "$RUNNER_TEMP/discrawl-native" --config "$CONFIG" init --db "$DB" --guild "$DISCRAWL_GUILD_ID" - sync_timeout=1200 - if [ "$RECONCILE_CLOUD" = "1" ]; then sync_timeout=300; fi - sync_args=(--guild "$DISCRAWL_GUILD_ID" --skip-members --latest-only --no-update) - if [ "$MEMBER_REFRESH" = "1" ]; then - sync_args=(--guild "$DISCRAWL_GUILD_ID" --with-members --latest-only --no-update) - fi - python3 scripts/private_discord_checkpoint.py run --timeout "$sync_timeout" \ - --writer-receipt "$RUNNER_TEMP/discrawl/sync.json" -- \ - "$RUNNER_TEMP/discrawl-native" --config "$CONFIG" sync "${sync_args[@]}" - - - name: Reconcile requested native source - timeout-minutes: 16 - id: reconcile - if: ${{ steps.sync.outcome == 'success' && github.event_name == 'workflow_dispatch' && inputs.reconcile_cloud }} - continue-on-error: true - env: - DISCORD_BOT_TOKEN: ${{ secrets.DISCORD_BOT_TOKEN }} - DISCRAWL_GUILD_ID: ${{ secrets.DISCRAWL_GUILD_ID }} - DISCRAWL_CLOUD_RECONCILE_PLAN: ${{ secrets.DISCRAWL_CLOUD_RECONCILE_PLAN }} - DISCRAWL_CLOUD_ARCHIVE: ${{ secrets.DISCRAWL_CLOUD_ARCHIVE }} - run: | - python3 scripts/private_discord_checkpoint.py run --timeout 900 \ - --writer-receipt "$RUNNER_TEMP/discrawl/reconcile.json" -- "$RUNNER_TEMP/discrawl-reconcile" + python3 scripts/private_discord_checkpoint.py acquire --timeout 1200 \ + --writer-receipt "$RUNNER_TEMP/discrawl/sync.json" \ + --native "$RUNNER_TEMP/discrawl-native" --config "$CONFIG" \ + --policy "$ACQUISITION_POLICY" --policy-sha256 "$POLICY_SHA256" --since "$HISTORY_SINCE" - name: Verify private ingestion checkpoint timeout-minutes: 15 @@ -110,14 +112,13 @@ jobs: env: DISCORD_BACKUP_TOKEN: ${{ secrets.DISCORD_BACKUP_TOKEN }} DISCRAWL_BACKUP_REPOSITORY: ${{ secrets.DISCRAWL_BACKUP_REPOSITORY }} - WRITER_RECEIPT: ${{ steps.reconcile.outcome != 'skipped' && 'reconcile.json' || 'sync.json' }} run: | python3 scripts/private_discord_checkpoint.py save --db "$DB" --receipt "$CHECKPOINT_RECEIPT" \ - --writer-receipt "$RUNNER_TEMP/discrawl/$WRITER_RECEIPT" + --writer-receipt "$RUNNER_TEMP/discrawl/sync.json" - name: Require successful native ingestion timeout-minutes: 1 - if: ${{ always() && !cancelled() && (steps.sync.outcome != 'success' || steps.reconcile.outcome == 'failure') }} + if: ${{ always() && !cancelled() && steps.sync.outcome != 'success' }} run: | echo "::error::Native ingestion did not complete. Publication was not requested; check the private checkpoint step for retained progress." exit 1 @@ -150,31 +151,3 @@ jobs: --message "sync: discord archive" --push >/dev/null 2>&1 echo "published=true" >> "$GITHUB_OUTPUT" echo "Archive publication completed." - - - name: Setup Node for cloud publication - timeout-minutes: 3 - if: ${{ vars.DISCRAWL_CLOUD_PUBLISH_ENABLED == '1' && steps.git-publish.outputs.published == 'true' }} - uses: actions/setup-node@949feb2413d6458794dcd2491c4babbbce0c15c1 # v7.1.0 - with: - node-version: '24' - - - name: Publish the same current state to D1 and R2 - if: ${{ vars.DISCRAWL_CLOUD_PUBLISH_ENABLED == '1' && steps.git-publish.outputs.published == 'true' }} - timeout-minutes: 40 - env: - DISCRAWL_CLOUD_ENDPOINT: ${{ secrets.DISCRAWL_CLOUD_ENDPOINT }} - DISCRAWL_CLOUD_ARCHIVE: ${{ secrets.DISCRAWL_CLOUD_ARCHIVE }} - DISCRAWL_CLOUD_AUTH_GITHUB_TOKEN: ${{ secrets.DISCORD_BACKUP_TOKEN }} - DISCRAWL_CLOUD_ACCESS_CLIENT_ID: ${{ secrets.DISCRAWL_CLOUD_ACCESS_CLIENT_ID }} - DISCRAWL_CLOUD_ACCESS_CLIENT_SECRET: ${{ secrets.DISCRAWL_CLOUD_ACCESS_CLIENT_SECRET }} - DISCRAWL_CLOUD_SNAPSHOT: ${{ runner.temp }}/discrawl/cloud.db - DISCRAWL_CLOUD_SOURCE_DB: ${{ github.workspace }}/.discrawl-ci/discrawl.db - run: | - umask 077 - trap 'rm -f "$DISCRAWL_CLOUD_SNAPSHOT"' EXIT - trap 'echo "::error::Cloud publication failed after archive publication."; exit 1' ERR - DISCRAWL_CLOUD_SOURCE_COMMIT="$(git -C "$BACKUP_REPO" rev-parse HEAD 2>/dev/null)" - export DISCRAWL_CLOUD_SOURCE_COMMIT - "$RUNNER_TEMP/discrawl-native" --config "$CONFIG" cloud publish --export-only "$DISCRAWL_CLOUD_SNAPSHOT" >/dev/null 2>&1 - node scripts/publish-cloud.mjs >/dev/null 2>&1 - echo "Cloud publication completed." diff --git a/CHANGELOG.md b/CHANGELOG.md index 07ea934d..6c6d596c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased +- Add manual, privately scoped report-history acquisition after the normal guild refresh, sharing one time budget and preserving failed progress in private checkpoints; hold recurring and cloud publication until separately qualified. - Identify private checkpoint API failures without exposing private output, and inspect a retained failed migration through a bounded read-only diagnostic. - Preserve complete native Discord crawl state in verified private checkpoints, with atomic restore, bounded retention, and an explicit migration from the retired workflow cache transport. diff --git a/docs/guides/private-checkpoints.md b/docs/guides/private-checkpoints.md index 27129737..82b75bd0 100644 --- a/docs/guides/private-checkpoints.md +++ b/docs/guides/private-checkpoints.md @@ -22,7 +22,7 @@ process group to disappear before issuing its completion receipt. Only then can the checkpoint helper open a read transaction and copy SQLite through its backup API. The pinned image supplies schema, table counts and integrity results. -The checkpoint is the **native ingestion boundary before Git/cloud publication**. +The checkpoint is the **native ingestion boundary before Git publication**. It deliberately precedes the publisher's later `share:*` bookkeeping. Replaying that boundary preserves messages, native cursors and failure history. Every workflow sets `DISCRAWL_NO_AUTO_UPDATE=1`; sync also uses `--no-update`. The daily @@ -36,7 +36,7 @@ Workflow concurrency provides serialization; GitHub's latest update is not a server-side compare-and-swap operation. Normal ingestion failure or the supervisor's time limit can still save committed -partial progress, but the job fails and skips Git/cloud publication. A forced +partial progress, but the job fails and skips Git publication. A forced runner cancellation or loss cannot guarantee a new upload. The previously verified checkpoint remains the recovery point until a new one is promoted. The reporter visibly skips a checkpoint unless its successful producer tuple @@ -58,13 +58,72 @@ the next ingestion until that same deletion succeeds. An interrupted draft also blocks new ingestion; an operator must reconcile that exact attempt. Do not bulk-delete releases by age or prefix. -The publisher has a 125-minute job ceiling. Phase caps total at most 120 minutes: -checkout/setup/build/configuration 18, restore 10, ingestion/reap 23, checkpoint -15, outcome gate 1, Git publication 10, Node setup 3 and cloud publication 40. -Normal sync gets 20 minutes; optional reconciliation shares that allowance with -a five-minute sync and a fifteen-minute reconciliation. This ceiling does not +The publisher has a 125-minute job ceiling. Phase caps total at most 79 minutes: +checkout/setup/build/configuration 18, policy validation 2, restore 10, +ingestion/reap 23, checkpoint 15, outcome gate 1 and Git publication 10. +The two acquisition phases share one 20-minute deadline. This ceiling does not promise a fifteen-minute completion time or durable FIFO scheduling. +## Private bounded acquisition + +The publisher currently has **only a manual trigger**. Keep it disabled until +the full-state migration is verified. Enabling it later permits a measured +manual run without also starting a cron schedule. Recurring acquisition and the +daily member refresh need a later reviewed cadence change, based on measured +restore, acquisition, checkpoint and publication durations. + +Each manual run requires `policy_commit` and `policy_sha256`. The existing +backup credential reads `policy/report-selectors.json` from the private backup +repository. Its `refs/heads/maintainer-report-policy` must point to exactly the +requested commit; the file is then fetched by that commit and its bytes must +match the requested SHA-256. A missing, changed or invalid policy stops before +native state is restored. No policy branch, repository variable or credential +is created by the workflow. + +The policy is a JSON object with these exact fields: + +- `schema`: integer `1`. +- `policy`: `repository`, full `revision`, repository-relative `path` and + `sha256` identifying the reviewed canonical report selector source. +- `catalog`: `archive_commit` and `manifest_sha256` identifying the catalog used + to select the roots and retained children. +- `guild_id`: the exact configured Discord guild ID. +- `message_root_ids` and `forum_root_ids`: arrays of native decimal IDs. +- `retained_threads`: objects with `id` and `parent_id`; each parent must be one + of the declared message or forum roots. Retained children of text roots must + be explicit targets: native targeted discovery expands forum children only. + +The approved policy digest binds that provenance and scope together. The file +is limited to 64 KiB and 4,096 distinct targets in total. It cannot contain +commands, arbitrary flags, wildcards, names or additional fields. Creation and +review of this private policy are separate operator actions. Do not store it on +archive `main`, in SQLite, or in public workflow inputs, logs or artifacts. The +workflow keeps its temporary copy private and outside all export paths. + +The supervisor first runs the existing guild-wide latest-only refresh. It then +runs native `sync --full --since` on the policy targets, starting at the previous +UTC midnight by default. An optional `since` input may pin a UTC midnight less +than 48 hours ago. It is fixed before restore and revalidated before acquisition; +a stale or future value stops rather than silently selecting a different day. +The second phase uses `--skip-members`, `--with-media=false` and `--no-update`. +Both phases share the same 1,200-second deadline, and the second +starts only after the first process group has exited. Any failure prevents Git +publication; committed partial work can still enter a verified private +checkpoint after all owned processes are reaped. + +This policy limits **acquisition cost and scope**, not archive privacy. Native +forum selection includes all bot-visible children of the selected forums, +including newly discovered children. Full forum discovery traverses archived +thread metadata even though message acquisition is date-bounded. The normal +Discord permissions and native unavailable markers still apply. A successful +process does not mean every channel is available or complete; report projection +must inspect each channel's bounded receipts and preserve its privacy rules. + +This publisher does not export or upload to cloud storage. Existing cloud data +is unchanged and is not made current by this private-only run. Restoring cloud +publication requires separate verification of its configured destination and +access policy. + ## One-time migration Keep the publisher, daily reporter and retired cache-repair workflow disabled. diff --git a/scripts/private_checkpoint_workflow_test.go b/scripts/private_checkpoint_workflow_test.go index 0a3873df..5af65780 100644 --- a/scripts/private_checkpoint_workflow_test.go +++ b/scripts/private_checkpoint_workflow_test.go @@ -56,8 +56,13 @@ func TestPrivateCheckpointWorkflowBoundaries(t *testing.T) { require.Contains(t, text[checkpoint:guard], "always() && !cancelled()") require.Contains(t, text[checkpoint:guard], "--writer-receipt") require.Contains(t, text[publish:], "success() && steps.checkpoint.outcome == 'success'") - require.Contains(t, text[publish:], "steps.git-publish.outputs.published == 'true'") - require.Contains(t, text, "--no-update") + require.Contains(t, text, "private_discord_checkpoint.py acquire --timeout 1200") + require.Contains(t, text, "private_discord_checkpoint.py policy") + require.NotContains(t, text, "schedule:") + require.NotContains(t, text, "cloud publish") + require.NotContains(t, text, "publish-cloud.mjs") + require.NotContains(t, text, "DISCRAWL_CLOUD_") + require.NotContains(t, text, "reconcile_cloud") require.NotContains(t, text, "actions/cache/restore") case "discord-backup-report": require.NotContains(t, text, " subscribe ") diff --git a/scripts/private_discord_checkpoint.py b/scripts/private_discord_checkpoint.py index dbf005fb..15d17814 100644 --- a/scripts/private_discord_checkpoint.py +++ b/scripts/private_discord_checkpoint.py @@ -2,7 +2,9 @@ """Private, complete native SQLite checkpoints. Never use the filtered share export.""" import argparse +import base64 import contextlib +from datetime import datetime, timedelta, timezone import gzip import hashlib import json @@ -30,6 +32,10 @@ BLOCK = 1024 * 1024 REQUIRED_TABLES = {"guilds", "channels", "members", "messages", "sync_state", "failure_ledger"} WORKFLOWS = {"publish-discord-backup.yml", "migrate-discord-checkpoint.yml"} +POLICY_REF = "refs/heads/maintainer-report-policy" +POLICY_PATH = "policy/report-selectors.json" +MAX_POLICY = 64 * 1024 +MAX_TARGETS = 4096 API_PHASES = {"api_read", "api_write", "repository_read", "archive_ref_read", "release_list", "latest_read", "release_read", "asset_list", "manifest_download", "part_download", "release_create", "asset_upload", "release_promote", "release_delete", "source_ref_read", "source_run_read", @@ -564,8 +570,95 @@ def report_ready(receipt, archive): return manifest["source"]["outcome"] == "success" and producer["producer"] == expected +def validate_policy(raw, expected_sha256, guild): + require(len(raw) <= MAX_POLICY and hash_value(expected_sha256) + and digest(raw) == expected_sha256, "policy_checksum") + policy = strict_json(raw) + keys(policy, "schema policy catalog guild_id message_root_ids forum_root_ids retained_threads") + require(type(policy["schema"]) is int and policy["schema"] == 1, "policy_schema") + source, catalog = policy["policy"], policy["catalog"] + keys(source, "repository revision path sha256") + keys(catalog, "archive_commit manifest_sha256") + commit = lambda value: isinstance(value, str) and re.fullmatch(r"[0-9a-f]{40}", value) + require(isinstance(source["repository"], str) + and re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", source["repository"]) + and commit(source["revision"]) and hash_value(source["sha256"]) + and isinstance(source["path"], str) and len(source["path"]) <= 512 + and re.fullmatch(r"[A-Za-z0-9_./-]+", source["path"]) + and all(part not in {"", ".", ".."} for part in source["path"].split("/")) + and commit(catalog["archive_commit"]) and hash_value(catalog["manifest_sha256"]), "policy_provenance") + native_id = lambda value: (isinstance(value, str) and re.fullmatch(r"[1-9][0-9]{0,19}", value) + and int(value) < 2 ** 64) + require(native_id(guild) and policy["guild_id"] == guild, "policy_guild") + targets = [] + for name in ("message_root_ids", "forum_root_ids"): + values = policy[name] + require(isinstance(values, list) and len(values) <= MAX_TARGETS + and all(native_id(value) for value in values), "policy_targets") + targets.extend(values) + parents = set(targets) + threads = policy["retained_threads"] + require(isinstance(threads, list) and len(threads) <= MAX_TARGETS, "policy_targets") + for thread in threads: + keys(thread, "id parent_id") + require(native_id(thread["id"]) and thread["parent_id"] in parents, "policy_thread_parent") + targets.append(thread["id"]) + require(0 < len(targets) <= MAX_TARGETS and len(set(targets)) == len(targets), "policy_targets") + return policy, sorted(targets, key=int) + + +def load_policy(github, revision, expected_sha256, guild, destination): + require(isinstance(revision, str) and re.fullmatch(r"[0-9a-f]{40}", revision) + and hash_value(expected_sha256) and not destination.exists(), "policy_pin") + ref = github.api("/git/ref/heads/maintainer-report-policy") + require(isinstance(ref, dict) and ref.get("ref") == POLICY_REF + and isinstance(ref.get("object"), dict) and ref["object"].get("type") == "commit" + and ref["object"].get("sha") == revision, "policy_ref") + blob = github.api("/contents/" + POLICY_PATH + "?ref=" + revision) + require(isinstance(blob, dict) and blob.get("type") == "file" and blob.get("path") == POLICY_PATH + and integer(blob.get("size"), MAX_POLICY) and blob.get("encoding") == "base64" + and isinstance(blob.get("content"), str), "policy_file") + raw = base64.b64decode(blob["content"].replace("\n", ""), validate=True) + require(len(raw) == blob["size"], "policy_size") + validate_policy(raw, expected_sha256, guild) + # This private input stays outside the archive, SQLite, and upload inventory. + with destination.open("xb") as file: + file.write(raw) + destination.chmod(0o600) + + +def acquisition_since(value=None): + now = datetime.now(timezone.utc) + if not value: + value = (now.replace(hour=0, minute=0, second=0, microsecond=0) + - timedelta(days=1)).isoformat().replace("+00:00", "Z") + require(isinstance(value, str) and re.fullmatch(r"\d{4}-\d{2}-\d{2}T00:00:00Z", value), "acquisition_window") + since = datetime.fromisoformat(value.replace("Z", "+00:00")) + require(timedelta() < now - since < timedelta(hours=48), "acquisition_window") + return value + + +def acquire(native, config, policy_path, expected_sha256, guild, completion, timeout=1200, since=None): + require(policy_path.is_file() and not policy_path.is_symlink() + and policy_path.stat().st_size <= MAX_POLICY, "policy_file") + _, targets = validate_policy(policy_path.read_bytes(), expected_sha256, guild) + require(0 < timeout <= 1200, "writer_arguments") + since = acquisition_since(since) + base = [str(native), "--config", str(config), "sync", "--source", "discord", "--guild", guild, + "--skip-members", "--no-update"] + # General archive refresh remains first; report roots only scope the extra + # bounded scan. Native forum expansion and report privacy keep their owners. + commands = [[*base, "--latest-only"], + [*base, "--full", "--since", since, "--channels", ",".join(targets), "--with-media=false"]] + return run_writers(commands, timeout, completion) + + def run_writer(command, timeout, completion=None): - require(command and timeout > 0, "writer_arguments") + return run_writers([command], timeout, completion) + + +def run_writers(commands, timeout, completion=None): + require(commands and all(commands) and timeout > 0, "writer_arguments") require(completion is None or not completion.exists(), "writer_receipt_exists") child = None cancelled = False @@ -580,36 +673,48 @@ def stop(_signal, _frame): try: with open(os.devnull, "wb") as output: env = {key: value for key, value in os.environ.items() if key not in {"GH_TOKEN", "DISCORD_BACKUP_TOKEN"}} - child = subprocess.Popen(command, stdout=output, stderr=output, env=env, start_new_session=True) deadline = time.monotonic() + timeout - while child.poll() is None and not cancelled and time.monotonic() < deadline: - try: - child.wait(timeout=0.2) - except subprocess.TimeoutExpired: - pass - if child.poll() is None: - cancelled = True - os.killpg(child.pid, signal.SIGTERM) + result = 1 + for command in commands: + if cancelled or time.monotonic() >= deadline: + cancelled = True + break try: - child.wait(timeout=5) - except subprocess.TimeoutExpired: + child = subprocess.Popen(command, stdout=output, stderr=output, env=env, start_new_session=True) + except OSError: + # Prior phases are reaped; preserve their committed progress. + result = 1 + break + while child.poll() is None and not cancelled and time.monotonic() < deadline: + try: + child.wait(timeout=0.2) + except subprocess.TimeoutExpired: + pass + if child.poll() is None: + cancelled = True + os.killpg(child.pid, signal.SIGTERM) + try: + child.wait(timeout=5) + except subprocess.TimeoutExpired: + os.killpg(child.pid, signal.SIGKILL) + child.wait(timeout=5) + # No next phase or checkpoint may run while descendants write. + with contextlib.suppress(ProcessLookupError): os.killpg(child.pid, signal.SIGKILL) - child.wait(timeout=5) - # A leader can exit while descendants still write. A completion - # receipt exists only after the whole task-owned group is gone. - with contextlib.suppress(ProcessLookupError): - os.killpg(child.pid, signal.SIGKILL) - deadline = time.monotonic() + 5 - while True: - try: - os.killpg(child.pid, 0) - except ProcessLookupError: + reap_deadline = time.monotonic() + 5 + while True: + try: + os.killpg(child.pid, 0) + except ProcessLookupError: + break + except PermissionError as error: + raise CheckpointError("writer_reap") from error + require(time.monotonic() < reap_deadline, "writer_reap") + time.sleep(0.05) + result = 1 if cancelled else child.returncode + if result != 0: break - except PermissionError as error: - raise CheckpointError("writer_reap") from error - require(time.monotonic() < deadline, "writer_reap") - time.sleep(0.05) - result = 1 if cancelled else child.returncode + result = 1 if cancelled else result if completion is not None: completion.write_bytes(canonical({"reaped": True, "outcome": "cancelled" if cancelled else "success" if result == 0 else "failure"})) @@ -622,7 +727,7 @@ def stop(_signal, _frame): def main(): os.umask(0o077) parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("operation", choices=("restore", "save", "migrate", "run", "report-check")) + parser.add_argument("operation", choices=("restore", "save", "migrate", "run", "report-check", "policy", "acquire")) parser.add_argument("--db", type=Path) parser.add_argument("--receipt", type=Path) parser.add_argument("--writer-receipt", type=Path) @@ -630,10 +735,27 @@ def main(): parser.add_argument("--timeout", type=int, default=2700) parser.add_argument("--prune", action="store_true") parser.add_argument("--archive", type=Path) + parser.add_argument("--policy", type=Path) + parser.add_argument("--policy-commit") + parser.add_argument("--policy-sha256") + parser.add_argument("--native", type=Path) + parser.add_argument("--config", type=Path) + parser.add_argument("--since") args, command = parser.parse_known_args() if args.operation == "run": return run_writer(command[1:] if command[:1] == ["--"] else command, args.timeout, args.writer_receipt) require(not command, "arguments") + if args.operation in {"policy", "acquire"}: + execution_context() + require(args.policy is not None, "arguments") + guild = os.environ.get("DISCRAWL_GUILD_ID", "") + if args.operation == "policy": + since = acquisition_since(args.since) + load_policy(GitHub(), args.policy_commit, args.policy_sha256, guild, args.policy) + print("since=" + since) + return 0 + require(args.native is not None and args.config is not None and args.writer_receipt is not None, "arguments") + return acquire(args.native, args.config, args.policy, args.policy_sha256, guild, args.writer_receipt, args.timeout, args.since) if args.operation == "report-check": require(args.receipt is not None and args.archive is not None, "arguments") return 0 if report_ready(args.receipt, args.archive) else 3 diff --git a/scripts/test_private_discord_checkpoint.py b/scripts/test_private_discord_checkpoint.py index 61d006d3..d682ac88 100644 --- a/scripts/test_private_discord_checkpoint.py +++ b/scripts/test_private_discord_checkpoint.py @@ -1,4 +1,5 @@ import copy +import base64 import contextlib import importlib.util import json @@ -327,7 +328,7 @@ def group(pid, sig): if sig == 0: return None return real_kill(pid, sig) - with patch.object(checkpoint.os, "killpg", group), patch.object(checkpoint.time, "monotonic", side_effect=[0, 0, 10, 20]): + with patch.object(checkpoint.os, "killpg", group), patch.object(checkpoint.time, "monotonic", side_effect=[0, 0, 0, 10, 20]): with self.assertRaisesRegex(checkpoint.CheckpointError, "writer_reap"): checkpoint.run_writer([sys.executable, "-c", "pass"], 1, completion) self.assertFalse(completion.exists()) @@ -353,6 +354,43 @@ def test_leader_exit_does_not_leave_a_writing_descendant(self): time.sleep(0.1) self.assertEqual(marker.read_bytes(), value) + def test_two_phase_failure_saves_progress_with_verified_parent(self): + parent = self.seed() + completion = self.root / "writer.json" + code = ("import sqlite3; d=sqlite3.connect(%r); " + "d.execute(\"INSERT INTO messages VALUES('phase-one','retained')\"); d.commit()") % str(self.path) + result = checkpoint.run_writers([[sys.executable, "-c", code], + [sys.executable, "-c", "raise SystemExit(9)"]], 3, completion) + self.assertEqual(result, 9) + self.assertEqual(json.loads(completion.read_bytes()), {"reaped": True, "outcome": "failure"}) + saved = checkpoint.save(self.github, self.path, self.receipt, "failure") + restored = self.root / "failed-acquisition.sqlite" + checkpoint.restore(self.github, restored, self.receipt) + with contextlib.closing(checkpoint.readonly(restored)) as db: + self.assertEqual(db.execute("SELECT value FROM messages WHERE id='phase-one'").fetchone(), ("retained",)) + self.assertEqual(set(self.github.releases), {parent["release_id"], saved["release_id"]}) + + def test_first_phase_failure_never_starts_second(self): + marker = self.root / "unexpected" + result = checkpoint.run_writers([[sys.executable, "-c", "raise SystemExit(7)"], + [sys.executable, "-c", "from pathlib import Path;Path(%r).touch()" % str(marker)]], 3) + self.assertEqual(result, 7) + self.assertFalse(marker.exists()) + + def test_two_phases_share_one_deadline(self): + completion = self.root / "writer.json" + result = checkpoint.run_writers([[sys.executable, "-c", "import time;time.sleep(.3)"], + [sys.executable, "-c", "import time;time.sleep(.4)"]], .5, completion) + self.assertNotEqual(result, 0) + self.assertEqual(json.loads(completion.read_bytes()), {"reaped": True, "outcome": "cancelled"}) + + def test_second_phase_spawn_failure_still_authorizes_retaining_first_phase(self): + completion = self.root / "writer.json" + result = checkpoint.run_writers([[sys.executable, "-c", "pass"], + [str(self.root / "missing-executable")]], 3, completion) + self.assertEqual(result, 1) + self.assertEqual(json.loads(completion.read_bytes()), {"reaped": True, "outcome": "failure"}) + class AdmissionTests(unittest.TestCase): def test_private_repository_identity_requires_no_admin_permission(self): @@ -393,5 +431,130 @@ def test_raw_api_failure_is_suppressed(self): self.assertNotIn(b"private API sentinel", result.stdout + result.stderr) +class AcquisitionPolicyTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.path = Path(self.temp.name) / "policy.json" + self.policy = {"schema": 1, "policy": {"repository": "example/reports", "revision": "a" * 40, + "path": "policy/channels.mjs", "sha256": "b" * 64}, + "catalog": {"archive_commit": "c" * 40, "manifest_sha256": "d" * 64}, + "guild_id": "100", "message_root_ids": ["11"], "forum_root_ids": ["12"], + "retained_threads": [{"id": "13", "parent_id": "12"}]} + + def validate(self, policy): + raw = checkpoint.canonical(policy) + return checkpoint.validate_policy(raw, checkpoint.digest(raw), "100") + + def test_policy_retains_provenance_and_builds_only_distinct_bounded_native_ids(self): + result, targets = self.validate(self.policy) + self.assertEqual(result, self.policy) + self.assertEqual(targets, ["11", "12", "13"]) + for mutate in [lambda p: p.update(commands=["untrusted"]), + lambda p: p.update(guild_id="101"), + lambda p: p["message_root_ids"].append("--all"), + lambda p: p["message_root_ids"].append(str(2 ** 64)), + lambda p: p["message_root_ids"].append("12"), + lambda p: p["message_root_ids"].extend(str(1000+i) for i in range(checkpoint.MAX_TARGETS)), + lambda p: p["retained_threads"][0].update(parent_id="14"), + lambda p: p["policy"].update(path="../private"), + lambda p: p["policy"].update(revision="main"), + lambda p: p["catalog"].update(manifest_sha256="wrong")]: + candidate = copy.deepcopy(self.policy) + mutate(candidate) + with self.subTest(candidate=candidate["guild_id"]), self.assertRaises(checkpoint.CheckpointError): + self.validate(candidate) + + def test_policy_fetch_pins_isolated_ref_exact_bytes_and_private_output(self): + raw = checkpoint.canonical(self.policy) + calls = [] + def api(suffix): + calls.append(suffix) + if suffix.startswith("/git/ref/"): + return {"ref": checkpoint.POLICY_REF, "object": {"type": "commit", "sha": "e" * 40}} + return {"type": "file", "path": checkpoint.POLICY_PATH, "encoding": "base64", + "size": len(raw), "content": base64.b64encode(raw).decode()} + github = SimpleNamespace(api=api) + checkpoint.load_policy(github, "e" * 40, checkpoint.digest(raw), "100", self.path) + self.assertEqual(self.path.read_bytes(), raw) + self.assertEqual(self.path.stat().st_mode & 0o777, 0o600) + self.assertEqual(calls, ["/git/ref/heads/maintainer-report-policy", + "/contents/policy/report-selectors.json?ref=" + "e" * 40]) + self.path.unlink() + with self.assertRaisesRegex(checkpoint.CheckpointError, "policy_ref"): + checkpoint.load_policy(github, "f" * 40, checkpoint.digest(raw), "100", self.path) + with self.assertRaisesRegex(checkpoint.CheckpointError, "policy_checksum"): + checkpoint.load_policy(github, "e" * 40, "f" * 64, "100", self.path) + self.assertFalse(self.path.exists()) + + def test_retained_text_children_are_explicit_targets_with_root_only_parents(self): + self.policy["retained_threads"].append({"id": "14", "parent_id": "11"}) + raw = checkpoint.canonical(self.policy) + self.path.write_bytes(raw) + with patch.object(checkpoint, "run_writers", return_value=0) as run: + self.assertEqual(checkpoint.acquire(Path("/native"), Path("/config"), self.path, + checkpoint.digest(raw), "100", Path("/receipt")), 0) + full = run.call_args.args[0][1] + self.assertEqual(full[full.index("--channels")+1], "11,12,13,14") + for parent in ["13", "15"]: + self.policy["retained_threads"][-1]["parent_id"] = parent + with self.subTest(parent=parent), self.assertRaisesRegex(checkpoint.CheckpointError, "policy_thread_parent"): + self.validate(self.policy) + + def test_acquisition_window_is_frozen_and_bounded(self): + class FrozenDatetime(checkpoint.datetime): + @classmethod + def now(cls, tz=None): + return cls(2026, 1, 3, tzinfo=checkpoint.timezone.utc) + with patch.object(checkpoint, "datetime", FrozenDatetime): + self.assertEqual(checkpoint.acquisition_since(), "2026-01-02T00:00:00Z") + self.assertEqual(checkpoint.acquisition_since("2026-01-02T00:00:00Z"), "2026-01-02T00:00:00Z") + for value in ["2026-01-01T00:00:00Z", "2026-01-03T00:00:00Z", "2026-01-04T00:00:00Z", + "2026-01-02T01:00:00Z", "2026-01-02T00:00:00+00:00", "yesterday"]: + with self.subTest(value=value), self.assertRaises(checkpoint.CheckpointError): + checkpoint.acquisition_since(value) + + def test_policy_cli_returns_only_frozen_date_and_acquire_reuses_it(self): + import io + frozen = "2026-01-02T00:00:00Z" + with patch.object(sys, "argv", ["checkpoint", "policy", "--policy", str(self.path), + "--policy-commit", "e" * 40, "--policy-sha256", "f" * 64]), \ + patch.object(checkpoint, "execution_context"), patch.object(checkpoint, "GitHub"), \ + patch.object(checkpoint, "acquisition_since", return_value=frozen), \ + patch.object(checkpoint, "load_policy"), patch("sys.stdout", new_callable=io.StringIO) as out: + self.assertEqual(checkpoint.main(), 0) + self.assertEqual(out.getvalue(), "since=" + frozen + "\n") + with patch.object(sys, "argv", ["checkpoint", "acquire", "--policy", str(self.path), + "--policy-sha256", "f" * 64, "--native", "/native", "--config", "/config", + "--writer-receipt", "/receipt", "--timeout", "1200", "--since", frozen]), \ + patch.object(checkpoint, "execution_context"), patch.object(checkpoint, "acquire", return_value=0) as run: + self.assertEqual(checkpoint.main(), 0) + self.assertEqual(run.call_args.args[-2:], (1200, frozen)) + + def test_acquisition_keeps_general_refresh_and_uses_previous_utc_midnight(self): + raw = checkpoint.canonical(self.policy) + self.path.write_bytes(raw) + class FrozenDatetime(checkpoint.datetime): + @classmethod + def now(cls, tz=None): + return cls(2026, 1, 3, 0, 1, tzinfo=checkpoint.timezone.utc) + with patch.object(checkpoint, "datetime", FrozenDatetime), patch.object(checkpoint, "run_writers", return_value=0) as run: + result = checkpoint.acquire(Path("/native"), Path("/config"), self.path, checkpoint.digest(raw), "100", Path("/receipt")) + self.assertEqual(result, 0) + commands, timeout, receipt = run.call_args.args + self.assertEqual((timeout, receipt), (1200, Path("/receipt"))) + self.assertIn("--latest-only", commands[0]) + self.assertNotIn("--channels", commands[0]) + self.assertEqual(commands[1][commands[1].index("--since")+1], "2026-01-02T00:00:00Z") + self.assertEqual(commands[1][commands[1].index("--channels")+1], "11,12,13") + self.assertIn("--full", commands[1]) + for command in commands: + self.assertIn("--no-update", command) + self.assertNotIn("--all", command) + self.assertNotIn("cloud", command) + with self.assertRaisesRegex(checkpoint.CheckpointError, "policy_checksum"): + checkpoint.acquire(Path("/native"), Path("/config"), self.path, "f" * 64, "100", Path("/receipt")) + + if __name__ == "__main__": unittest.main()