fix(ci): diagnose retained checkpoint readback failures

.github/workflows/migrate-discord-checkpoint.yml

@@ -3,20 +3,38 @@ name: migrate-discord-checkpoint
 on:
   workflow_dispatch:
     inputs:
+      mode:
+        description: Migrate once or inspect one retained failed migration without changes
+        type: choice
+        options: [migrate, diagnose]
+        default: migrate
+        required: true
       expected_sha:
         description: Reviewed main commit containing this migration
         type: string
         required: true
       producer_revision:
         description: Exact source revision of the successful cache producer
         type: string
-        required: true
+        required: false
       cache_identity:
         description: Exact reviewed cache id, key, ref, version, created_at and size_in_bytes JSON
         type: string
-        required: true
+        required: false
+      original_run:
+        description: Diagnose only - exact failed migration run
+        type: string
+        required: false
+      manifest_sha256:
+        description: Diagnose only - reviewed private manifest SHA-256
+        type: string
+        required: false
+      inventory_sha256:
+        description: Diagnose only - reviewed asset inventory SHA-256
+        type: string
+        required: false
       acknowledgement:
-        description: Type MIGRATE_FULL_NATIVE_STATE
+        description: Type MIGRATE_FULL_NATIVE_STATE or DIAGNOSE_EXISTING_CHECKPOINT for the selected mode
         type: string
         required: true
 
@@ -30,6 +48,7 @@ concurrency:
 
 jobs:
   migrate:
+    if: inputs.mode == 'migrate'
     runs-on: ubuntu-latest
     timeout-minutes: 45
     env:
@@ -83,3 +102,29 @@ jobs:
           python3 scripts/private_discord_checkpoint.py restore --db "$RUNNER_TEMP/independent-restore.sqlite" \
             --receipt "$RUNNER_TEMP/independent-restore.json"
           echo "Private full-state migration and independent restore verified; schedules remain unchanged."
+
+  diagnose:
+    if: inputs.mode == 'diagnose'
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    env:
+      DISCRAWL_NO_AUTO_UPDATE: "1"
+      CHECKPOINT_MODE: diagnose
+      EXPECTED_SHA: ${{ inputs.expected_sha }}
+      ORIGINAL_RUN: ${{ inputs.original_run }}
+      MANIFEST_SHA256: ${{ inputs.manifest_sha256 }}
+      INVENTORY_SHA256: ${{ inputs.inventory_sha256 }}
+      MIGRATION_ACK: ${{ inputs.acknowledgement }}
+    steps:
+      - name: Checkout reviewed diagnostic
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+
+      - name: Inspect retained draft and one existing part without changes
+        timeout-minutes: 16
+        env:
+          GH_TOKEN: ${{ github.token }}
+          DISCORD_BACKUP_TOKEN: ${{ secrets.DISCORD_BACKUP_TOKEN }}
+          DISCRAWL_BACKUP_REPOSITORY: ${{ secrets.DISCRAWL_BACKUP_REPOSITORY }}
+        run: |
+          test "$(git rev-parse HEAD)" = "$EXPECTED_SHA"
+          python3 scripts/verify_discord_checkpoint_migration.py

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## Unreleased
 
+- 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.
 
 - 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.

docs/guides/private-checkpoints.md

@@ -86,3 +86,35 @@ old caches. Keep schedules disabled until source restore/publication, ongoing
 acquisition coverage and measured end-to-end cadence are qualified. Source merge
 or a small fixture alone does not establish production credential access,
 full-state migration, or report completeness.
+
+## Diagnose a retained migration draft
+
+If migration fails, preserve its draft and recovery cache. Do not rerun the
+migration to diagnose an unknown API error. The same manual workflow offers
+`mode: diagnose`, with acknowledgement `DIAGNOSE_EXISTING_CHECKPOINT`, the
+reviewed current `expected_sha`, `original_run`, `manifest_sha256`, and
+`inventory_sha256`. The inventory hash covers canonical JSON of assets sorted
+by name, each containing `id`, `name`, `size`, and `digest`. Review these pins
+through authenticated metadata before dispatch; private repository and asset
+IDs are not workflow inputs.
+
+This mode requires the original manual migration's first attempt to have
+completed with failure, the current diagnostic to be a reviewed first attempt
+on main, and all archive writers to remain disabled and idle. It checks the
+existing draft, manifest, asset identities, private archive main, and absence
+of latest, downloads only the first manifest part, verifies its bytes, then
+rechecks those bindings. It never decompresses or restores a database, reads a
+cache, uploads, promotes, or deletes a release. Temporary downloads are removed.
+
+The diagnostic has one 15-minute deadline inside a 20-minute job. Metadata calls
+allow at most 60 seconds; the part download allows at most 600 seconds, always
+limited by the remaining deadline. API errors expose only a fixed operation,
+numeric exit or timeout, and an HTTP status when available. Private command
+output is drained with a bounded retained prefix and never logged. Public
+admission and private reads share this process boundary while retaining their
+separate tokens. The helper terminates its command group and allows up to five
+seconds to verify it has disappeared before inspecting output; unknown cleanup
+state stops the diagnostic. A successful
+probe proves the current credential and first-part read path; it does not locate
+an earlier unrecorded failure or verify the other parts. Full recovery still
+requires a separately reviewed complete restore and parity check.

scripts/private_checkpoint_workflow_test.go

@@ -68,7 +68,14 @@ func TestPrivateCheckpointWorkflowBoundaries(t *testing.T) {
 		case "migrate-discord-checkpoint":
 			require.NotContains(t, text, "schedule:")
 			require.Contains(t, text, "fail-on-cache-miss: true")
-			require.Equal(t, 2, strings.Count(text, "verify_discord_checkpoint_migration.py"))
+			require.Equal(t, 3, strings.Count(text, "verify_discord_checkpoint_migration.py"))
+			require.Contains(t, text, "if: inputs.mode == 'migrate'")
+			require.Contains(t, text, "if: inputs.mode == 'diagnose'")
+			diagnostic := text[strings.Index(text, "  diagnose:"):]
+			require.Contains(t, diagnostic, "timeout-minutes: 20")
+			require.Contains(t, diagnostic, "CHECKPOINT_MODE: diagnose")
+			require.NotContains(t, diagnostic, "actions/cache")
+			require.NotContains(t, diagnostic, "private_discord_checkpoint.py")
 			require.Contains(t, text, "independent-restore.sqlite")
 			require.NotContains(t, text, " sync ")
 			require.NotContains(t, text, "cloud publish")

scripts/private_discord_checkpoint.py

@@ -10,6 +10,7 @@
 from pathlib import Path
 import re
 import resource
+import selectors
 import shutil
 import signal
 import sqlite3
@@ -29,6 +30,10 @@
 BLOCK = 1024 * 1024
 REQUIRED_TABLES = {"guilds", "channels", "members", "messages", "sync_state", "failure_ledger"}
 WORKFLOWS = {"publish-discord-backup.yml", "migrate-discord-checkpoint.yml"}
+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",
+              "source_workflow_read", "source_active_runs", "source_cache_read"}
 
 
 class CheckpointError(Exception):
@@ -208,8 +213,94 @@ def unpack(manifest, directory, destination):
         require(database_facts(db) == {k: expected[k] for k in ("user_version", "schema_sha256", "tables")}, "restore_parity")
 
 
+def reap_command(child, phase):
+    # A reaped leader and closed stderr do not prove stdout is stable: a
+    # descendant can still own it. Unknown group state must not yield bytes.
+    deadline = time.monotonic() + 5
+    code = "checkpoint_api phase=" + phase + " exit=reap"
+    try:
+        os.killpg(child.pid, signal.SIGKILL)
+    except ProcessLookupError:
+        pass
+    except PermissionError as error:
+        raise CheckpointError(code) from error
+    try:
+        child.wait(timeout=max(0, deadline - time.monotonic()))
+    except subprocess.TimeoutExpired as error:
+        raise CheckpointError(code) from error
+    while True:
+        try:
+            os.killpg(child.pid, 0)
+        except ProcessLookupError:
+            return
+        except PermissionError as error:
+            raise CheckpointError(code) from error
+        require(time.monotonic() < deadline, code)
+        time.sleep(0.05)
+
+
+def github_command(arguments, env, phase, destination=None, timeout=600, missing=False,
+                   limit=MAX_MANIFEST, deadline=None):
+    require(phase in API_PHASES, "api_phase")
+    command_deadline = time.monotonic() + timeout
+    if deadline is not None:
+        command_deadline = min(command_deadline, deadline)
+    require(time.monotonic() < command_deadline, "checkpoint_api phase=" + phase + " exit=timeout")
+
+    def bound_output():
+        resource.setrlimit(resource.RLIMIT_FSIZE, (limit, limit))
+
+    with (destination.open("xb") if destination is not None else tempfile.TemporaryFile()) as output:
+        child = subprocess.Popen(["gh", *arguments], stdout=output, stderr=subprocess.PIPE,
+                                 env=env, preexec_fn=bound_output, start_new_session=True)
+        captured = bytearray()
+        failure = None
+        try:
+            # Drain even after the retained prefix fills: waiting on a full
+            # stderr pipe deadlocks, and inherited pipes need the same deadline.
+            with selectors.DefaultSelector() as selector:
+                os.set_blocking(child.stderr.fileno(), False)
+                selector.register(child.stderr, selectors.EVENT_READ)
+                while selector.get_map() or child.poll() is None:
+                    require(time.monotonic() < command_deadline, "checkpoint_api phase=" + phase + " exit=timeout")
+                    for key, _ in selector.select(min(0.1, max(0, command_deadline - time.monotonic()))):
+                        try:
+                            chunk = os.read(key.fd, 16384)
+                        except BlockingIOError:
+                            continue
+                        if not chunk:
+                            selector.unregister(key.fileobj)
+                        captured.extend(chunk[:max(0, 65536 - len(captured))])
+        except BaseException as error:
+            failure = error
+        finally:
+            try:
+                reap_command(child, phase)
+            finally:
+                child.stderr.close()
+        if failure is not None:
+            raise failure
+        size = output.tell()
+        output.seek(0)
+        raw = output.read(min(size, limit)) if destination is None else b""
+        if missing and child.returncode != 0:
+            if re.match(rb"HTTP/\S+ 404\b", raw):
+                return None
+        if child.returncode != 0:
+            status = re.search(rb"(?:^HTTP/\S+ |\(HTTP )([1-5][0-9]{2})(?:\b|\))", raw[:4096] + bytes(captured))
+            raise CheckpointError(f"checkpoint_api phase={phase} exit={child.returncode}"
+                                  + (" http=" + status[1].decode("ascii") if status else ""))
+        if destination is not None:
+            require(size <= limit, "asset_size")
+            return None
+        require(size <= limit, "checkpoint_api_size")
+        return raw
+
+
 class GitHub:
-    def __init__(self):
+    def __init__(self, deadline=None, read_only=False):
+        self.deadline = deadline
+        self.read_only = read_only
         self.repository = os.environ.get("DISCRAWL_BACKUP_REPOSITORY", "")
         require(re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", self.repository), "checkpoint_repository")
         self.base = "repos/" + self.repository
@@ -218,58 +309,63 @@ def __init__(self):
                 and repo.get("archived") is False and integer(repo.get("id")), "private_repository")
         self.identity = {"name": self.repository, "id": repo["id"]}
 
-    def command(self, arguments, destination=None, timeout=600, missing=False):
-        # The official CLI owns authentication/redirects. Neither its error body
-        # nor private asset paths are allowed into this public workflow's logs.
+    def command(self, arguments, destination=None, timeout=600, missing=False, phase="repository_read", limit=None):
+        # Private custody uses only its existing token. Public admission passes
+        # its own workflow token to the same bounded process owner.
         env = {**os.environ, "GH_TOKEN": os.environ.get("DISCORD_BACKUP_TOKEN", ""),
                "GH_PROMPT_DISABLED": "1", "GH_DEBUG": "", "GH_HOST": "github.com"}
         require(env["GH_TOKEN"], "checkpoint_credential")
-        limit = MAX_PART if destination is not None else MAX_MANIFEST
-
-        def bound_output():
-            resource.setrlimit(resource.RLIMIT_FSIZE, (limit, limit))
-
-        with (destination.open("xb") if destination is not None else tempfile.TemporaryFile()) as output:
-            result = subprocess.run(["gh", *arguments], stdout=output, stderr=subprocess.DEVNULL,
-                                    env=env, timeout=timeout, check=False, preexec_fn=bound_output)
-            size = output.tell()
-            output.seek(0)
-            if missing and result.returncode != 0:
-                raw = output.read()
-                if re.match(rb"HTTP/\S+ 404\b", raw):
-                    return None
-            require(result.returncode == 0, "checkpoint_api")
-            if destination is not None:
-                require(size <= MAX_PART, "asset_size")
-                return None
-            require(size <= MAX_MANIFEST, "checkpoint_api_size")
-            return output.read()
+        require(not self.read_only or (arguments[0] == "api" and "--method" not in arguments
+                                      and "--input" not in arguments), "diagnostic_read_only")
+        if self.read_only:
+            arguments = [*arguments, "--method", "GET"]
+        limit = limit if limit is not None else MAX_PART if destination is not None else MAX_MANIFEST
+        return github_command(arguments, env, phase, destination, timeout, missing, limit, self.deadline)
 
     def api(self, suffix, data=None, missing=False):
+        if suffix == "":
+            phase = "repository_read"
+        elif suffix == "/git/ref/heads/main":
+            phase = "archive_ref_read"
+        elif suffix.startswith("/releases?"):
+            phase = "release_list"
+        elif suffix == "/releases/latest":
+            phase = "latest_read"
+        elif re.fullmatch(r"/releases/[1-9][0-9]*/assets\?per_page=100", suffix):
+            phase = "asset_list"
+        elif re.fullmatch(r"/releases/[1-9][0-9]*", suffix):
+            phase = "release_promote" if data is not None else "release_read"
+        else:
+            phase = "api_write" if data is not None else "api_read"
         args = ["api", self.base + suffix, "--include", "-H", "X-GitHub-Api-Version: 2022-11-28"]
         def parse(raw):
             if raw is None:
                 return None
             parts = re.split(rb"\r?\n\r?\n", raw, maxsplit=1)
-            require(len(parts) == 2 and re.match(rb"HTTP/\S+ 2[0-9][0-9]\b", parts[0]), "checkpoint_api_status")
+            require(len(parts) == 2 and re.match(rb"HTTP/\S+ 2[0-9][0-9]\b", parts[0]), "checkpoint_api_status phase=" + phase)
             return strict_json(parts[1])
         if data is None:
-            return parse(self.command(args, missing=missing))
+            return parse(self.command(args, missing=missing, phase=phase, timeout=60 if self.read_only else 600))
         with tempfile.NamedTemporaryFile(mode="wb") as body:
             body.write(canonical(data))
             body.flush()
-            return parse(self.command([*args, "--method", "PATCH", "--input", body.name]))
+            return parse(self.command([*args, "--method", "PATCH", "--input", body.name], phase=phase))
 
     def create(self, tag, target):
         with tempfile.NamedTemporaryFile(mode="wb") as body:
             body.write(canonical({"tag_name": tag, "target_commitish": target, "name": tag,
                                   "draft": True, "prerelease": False, "make_latest": "false"}))
             body.flush()
-            return strict_json(self.command(["api", self.base + "/releases", "--method", "POST", "--input", body.name]))
+            return strict_json(self.command(["api", self.base + "/releases", "--method", "POST", "--input", body.name],
+                                            phase="release_create"))
 
     def download(self, asset, destination):
+        manifest = asset["name"] == MANIFEST
         self.command(["api", self.base + "/releases/assets/" + str(asset["id"]),
-                      "-H", "Accept: application/octet-stream"], destination)
+                      "-H", "Accept: application/octet-stream"], destination,
+                     timeout=60 if self.read_only and manifest else 600,
+                     phase="manifest_download" if manifest else "part_download",
+                     limit=MAX_MANIFEST if manifest else MAX_PART)
 
 
 def release_manifest(github, release, directory, draft=False):
@@ -386,7 +482,7 @@ def retire_grandparent(github, current, directory):
             _, _, actual = release_manifest(github, candidate, Path(temp))
         require(actual == grandparent, "retention_lineage")
         require(github.api("/releases/latest")["id"] == current["reference"]["release_id"], "parent_changed")
-        github.command(["api", github.base + '/releases/' + str(grandparent["release_id"]), "--method", "DELETE"])
+        github.command(["api", github.base + '/releases/' + str(grandparent["release_id"]), "--method", "DELETE"], phase="release_delete")
         require(github.api('/releases/' + str(grandparent["release_id"]), missing=True) is None, "retirement_unverified")
 
 
@@ -425,7 +521,7 @@ def save(github, database, receipt, outcome, migration=None):
         draft = github.create(tag, archive_commit)
         require(draft.get("draft") is True and draft.get("tag_name") == tag and integer(draft.get("id")), "draft_identity")
         github.command(["release", "upload", tag, "--repo", github.repository,
-                        str(directory / MANIFEST), *(str(directory / part["name"]) for part in manifest["parts"])])
+                        str(directory / MANIFEST), *(str(directory / part["name"]) for part in manifest["parts"])], phase="asset_upload")
         standalone.unlink()
         for part in manifest["parts"]:
             (directory / part["name"]).unlink()

scripts/test_private_discord_checkpoint.py

@@ -92,7 +92,7 @@ def create(self, tag, target):
         self.releases[release["id"]] = release
         return copy.deepcopy(release)
 
-    def command(self, args):
+    def command(self, args, **_options):
         if args[:2] == ["release", "upload"]:
             if self.fail_upload:
                 raise checkpoint.CheckpointError("checkpoint_api")
@@ -389,7 +389,7 @@ def test_raw_api_failure_is_suppressed(self):
             result = subprocess.run([sys.executable, str(SCRIPT), "restore", "--db", temp + "/state.sqlite",
                                      "--receipt", temp + "/receipt.json"], capture_output=True, env=env)
             self.assertEqual(result.returncode, 1)
-            self.assertEqual(result.stderr, b"checkpoint failed: checkpoint_api\n")
+            self.assertEqual(result.stderr, b"checkpoint failed: checkpoint_api phase=repository_read exit=1\n")
             self.assertNotIn(b"private API sentinel", result.stdout + result.stderr)