From 93020fe74bef0271aea39dc0ff8a0100bb2ead6d Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sat, 17 Oct 2026 22:12:43 +0800 Subject: [PATCH 1/3] fix: refresh private report history within one ingestion budget --- .github/workflows/publish-discord-backup.yml | 105 ++++------- CHANGELOG.md | 2 + docs/guides/private-checkpoints.md | 72 +++++++- scripts/private_checkpoint_workflow_test.go | 9 +- scripts/private_discord_checkpoint.py | 177 ++++++++++++++++--- scripts/test_private_discord_checkpoint.py | 151 +++++++++++++++- 6 files changed, 412 insertions(+), 104 deletions(-) diff --git a/.github/workflows/publish-discord-backup.yml b/.github/workflows/publish-discord-backup.yml index b1d37cff..8c41b542 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 within 48 hours; 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 eaf6e885..c9c2e2be 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## 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. + - 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. - Export current history-permission denials alongside retained window receipts, fail the sync if that status cannot be saved, and clear the unavailable marker after a successful full or targeted retry. diff --git a/docs/guides/private-checkpoints.md b/docs/guides/private-checkpoints.md index 59032fba..f6a957f4 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,71 @@ 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 forum roots. + +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 within +the last 48 hours. 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 f99f2cb2..ee6798ea 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 45ea7ea0..4b395c79 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 @@ -29,6 +31,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 class CheckpointError(Exception): @@ -468,8 +474,94 @@ 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) + 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 policy["forum_root_ids"], "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 @@ -484,36 +576,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"})) @@ -526,7 +630,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) @@ -534,10 +638,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 f5c6a5b6..a1abd357 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,116 @@ 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_acquisition_window_is_frozen_and_bounded(self): + 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): + self.assertEqual(checkpoint.acquisition_since(), "2026-01-02T00:00:00Z") + self.assertEqual(checkpoint.acquisition_since("2026-01-03T00:00:00Z"), "2026-01-03T00:00:00Z") + for value in ["2026-01-01T00: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() From 18429cb14c0e0ac7ce349750b14f301ef232771d Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sat, 17 Oct 2026 22:16:14 +0800 Subject: [PATCH 2/3] fix: enforce strict history acquisition window bound --- .github/workflows/publish-discord-backup.yml | 2 +- docs/guides/private-checkpoints.md | 4 ++-- scripts/private_discord_checkpoint.py | 2 +- scripts/test_private_discord_checkpoint.py | 6 +++--- 4 files changed, 7 insertions(+), 7 deletions(-) diff --git a/.github/workflows/publish-discord-backup.yml b/.github/workflows/publish-discord-backup.yml index 8c41b542..69c2b4f0 100644 --- a/.github/workflows/publish-discord-backup.yml +++ b/.github/workflows/publish-discord-backup.yml @@ -13,7 +13,7 @@ on: type: string required: true since: - description: Optional UTC midnight within 48 hours; defaults to previous UTC midnight + description: Optional UTC midnight less than 48 hours ago; defaults to previous UTC midnight type: string required: false diff --git a/docs/guides/private-checkpoints.md b/docs/guides/private-checkpoints.md index f6a957f4..6f82a548 100644 --- a/docs/guides/private-checkpoints.md +++ b/docs/guides/private-checkpoints.md @@ -101,8 +101,8 @@ 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 within -the last 48 hours. It is fixed before restore and revalidated before acquisition; +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 diff --git a/scripts/private_discord_checkpoint.py b/scripts/private_discord_checkpoint.py index 4b395c79..0340112c 100644 --- a/scripts/private_discord_checkpoint.py +++ b/scripts/private_discord_checkpoint.py @@ -537,7 +537,7 @@ def acquisition_since(value=None): - 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") + require(timedelta() < now - since < timedelta(hours=48), "acquisition_window") return value diff --git a/scripts/test_private_discord_checkpoint.py b/scripts/test_private_discord_checkpoint.py index a1abd357..6039d27c 100644 --- a/scripts/test_private_discord_checkpoint.py +++ b/scripts/test_private_discord_checkpoint.py @@ -491,11 +491,11 @@ def test_acquisition_window_is_frozen_and_bounded(self): class FrozenDatetime(checkpoint.datetime): @classmethod def now(cls, tz=None): - return cls(2026, 1, 3, 0, 1, tzinfo=checkpoint.timezone.utc) + 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-03T00:00:00Z"), "2026-01-03T00:00:00Z") - for value in ["2026-01-01T00:00:00Z", "2026-01-04T00: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) From f07e2ef2487bf189e6c494fd85c35b9b3cfb838e Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sat, 17 Oct 2026 22:29:08 +0800 Subject: [PATCH 3/3] fix: include retained threads under selected message roots --- docs/guides/private-checkpoints.md | 3 ++- scripts/private_discord_checkpoint.py | 3 ++- scripts/test_private_discord_checkpoint.py | 14 ++++++++++++++ 3 files changed, 18 insertions(+), 2 deletions(-) diff --git a/docs/guides/private-checkpoints.md b/docs/guides/private-checkpoints.md index 6f82a548..7d0e64ad 100644 --- a/docs/guides/private-checkpoints.md +++ b/docs/guides/private-checkpoints.md @@ -90,7 +90,8 @@ The policy is a JSON object with these exact fields: - `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 forum roots. + 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 diff --git a/scripts/private_discord_checkpoint.py b/scripts/private_discord_checkpoint.py index 0340112c..19e62d21 100644 --- a/scripts/private_discord_checkpoint.py +++ b/scripts/private_discord_checkpoint.py @@ -500,11 +500,12 @@ def validate_policy(raw, expected_sha256, guild): 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 policy["forum_root_ids"], "policy_thread_parent") + 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) diff --git a/scripts/test_private_discord_checkpoint.py b/scripts/test_private_discord_checkpoint.py index 6039d27c..88622c8c 100644 --- a/scripts/test_private_discord_checkpoint.py +++ b/scripts/test_private_discord_checkpoint.py @@ -487,6 +487,20 @@ def api(suffix): 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