Implement bounded primary archive transfer and explicit essentials capture

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883

scripts/capture_forgejo_archive.py

@@ -9,18 +9,24 @@ import time
 import zipfile
 
 
-def validate_archive(path):
+ESSENTIALS_FLAGS = ['--skip-package-data', '--skip-log', '--skip-index', '--skip-repo-archives']
+
+
+def validate_archive(path, profile="full"):
     with zipfile.ZipFile(path) as archive:
         if sum(i.file_size for i in archive.infolist()) > 20*1024**3:
             raise ValueError('archive exceeds recovery size bound')
         names = set(archive.namelist())
         if 'forgejo-db.sql' not in names or not any(n.startswith('repos/') for n in names):
             raise ValueError('required archive content missing')
+        if profile == 'essentials' and any(n.startswith(('data/packages/', 'data/repo-archive/', 'data/indexers/')) and not n.endswith('/') for n in names):
+            raise ValueError('excluded data present in essentials archive')
         if archive.testzip() is not None:
             raise ValueError('archive checksum failure')
 
 
-def capture(namespace, pod, destination):
+def capture(namespace, pod, destination, profile="full"):
+    if profile not in ("full", "essentials"): raise ValueError("unknown archive profile")
     k = ['kubectl', 'exec', '--request-timeout=120s', '-n', namespace, pod, '-c', 'gitea', '--']
     def call(args, timeout=150):
         r = subprocess.run(k + args, capture_output=True, timeout=timeout)
@@ -30,7 +36,8 @@ def capture(namespace, pod, destination):
     completed = False
     try:
         # Completion belongs to this exact process, not a global pgrep or file existence.
-        command = f'umask 077; forgejo dump -f {remote}.zip >{remote}.log 2>&1; result=$?; printf "%s" "$result" >{remote}.exit'
+        flags = ' '.join(ESSENTIALS_FLAGS) if profile == 'essentials' else ''
+        command = f'umask 077; forgejo dump {flags} -f {remote}.zip >{remote}.log 2>&1; result=$?; printf "%s" "$result" >{remote}.exit'
         call(['sh','-c', 'nohup sh -c "$1" >/dev/null 2>&1 </dev/null &', 'sh', command])
         for _ in range(180):
             r = subprocess.run(k + ['cat',remote+'.exit'], capture_output=True, timeout=30)
@@ -53,7 +60,9 @@ def capture(namespace, pod, destination):
         with destination.open('rb') as source:
             actual = hashlib.file_digest(source,'sha256').hexdigest()
         if actual != expected: raise ValueError('archive transfer mismatch')
-        validate_archive(destination)
+        validate_archive(destination, profile)
+        if profile == "essentials" and destination.stat().st_size > 600*1024**2:
+            raise ValueError("essentials archive exceeds budget")
     finally:
         # Do not remove a file while a timed-out producer may still be writing it.
         if completed:
@@ -64,8 +73,9 @@ def main():
     p=argparse.ArgumentParser(description=__doc__)
     p.add_argument('--namespace',required=True); p.add_argument('--pod',required=True)
     p.add_argument('--output',required=True,type=Path)
+    p.add_argument("--profile",choices=["full","essentials"],default="full")
     a=p.parse_args()
-    try: capture(a.namespace,a.pod,a.output)
+    try: capture(a.namespace,a.pod,a.output,a.profile)
     except Exception:
         print('ERROR: Forgejo archive capture or integrity validation failed')
         return 1

scripts/scaleway_forgejo_archive.py

@@ -0,0 +1,99 @@
+#!/usr/bin/env python3
+"""Bounded primary archive multipart PUT/GET using existing backup custody."""
+import argparse
+import base64
+from datetime import datetime, timezone
+import hashlib
+import json
+import os
+from pathlib import Path
+import secrets
+import subprocess
+from state_hub_preflight_lane import assert_cluster
+
+ENDPOINT='https://s3.nl-ams.scw.cloud'
+BUCKET='railiance-platform-pg-backup'
+PREFIX='platform-pg/application-archives/forgejo/'
+PART_SIZE=64*1024**2
+
+
+def digest(path):
+    with path.open('rb') as stream:
+        return hashlib.file_digest(stream,'sha256').hexdigest()
+
+
+def transfer(client, source, output, receipt, checkpoint=lambda: None):
+    expected=digest(source)
+    key=PREFIX+datetime.now(timezone.utc).strftime('%Y/%m/%d/%H%M%S-')+secrets.token_hex(12)+'.zip.age'
+    receipt.update(destination='s3://'+BUCKET+'/'+key, ciphertext_sha256=expected,
+                   ciphertext_bytes=source.stat().st_size, stage='multipart_upload', uploaded_bytes=0)
+    checkpoint()
+    upload=None
+    try:
+        upload=client.create_multipart_upload(Bucket=BUCKET,Key=key,ContentType='application/octet-stream',Metadata={'sha256':expected})['UploadId']
+        parts=[]
+        with source.open('rb') as stream:
+            while block:=stream.read(PART_SIZE):
+                number=len(parts)+1
+                result=client.upload_part(Bucket=BUCKET,Key=key,UploadId=upload,PartNumber=number,Body=block)
+                parts.append({'PartNumber':number,'ETag':result['ETag']})
+                receipt['uploaded_bytes']+=len(block);checkpoint()
+        completed=client.complete_multipart_upload(Bucket=BUCKET,Key=key,UploadId=upload,MultipartUpload={'Parts':parts})
+        upload=None
+        version=completed.get('VersionId')
+        receipt.update(stage='primary_download',multipart_completed=True,version_pinned=bool(version),downloaded_bytes=0)
+        checkpoint()
+        args={'Bucket':BUCKET,'Key':key}
+        if version: args['VersionId']=version
+        response=client.get_object(**args)
+        if response['ContentLength']!=source.stat().st_size: raise ValueError('primary_length_mismatch')
+        with response['Body'] as body, output.open('xb') as target:
+            output.chmod(0o600)
+            while block:=body.read(8*1024**2):
+                target.write(block);receipt['downloaded_bytes']+=len(block);checkpoint()
+        if digest(output)!=expected: raise ValueError('primary_digest_mismatch')
+        receipt.update(status='primary_fetched_pending_application_restore',stage='transfer_verified',download_hash_matches=True)
+        checkpoint()
+    finally:
+        if upload:
+            try:
+                client.abort_multipart_upload(Bucket=BUCKET,Key=key,UploadId=upload)
+                receipt['multipart_aborted']=True
+            except Exception:
+                receipt['multipart_aborted']=False
+            checkpoint()
+
+
+def main():
+    p=argparse.ArgumentParser(description=__doc__)
+    p.add_argument('--source',required=True,type=Path);p.add_argument('--source-receipt',required=True,type=Path)
+    p.add_argument('--output',required=True,type=Path);p.add_argument('--receipt',required=True,type=Path)
+    p.add_argument('--kubeconfig',required=True)
+    a=p.parse_args()
+    receipt={'schema':'platform.forgejo-primary-archive.v1','status':'running'}
+    fd=os.open(a.receipt,os.O_CREAT|os.O_EXCL|os.O_WRONLY,0o600);os.close(fd)
+    def checkpoint(): a.receipt.write_text(json.dumps(receipt,indent=2)+'\n')
+    try:
+        source=json.loads(a.source_receipt.read_text())
+        if (source.get('status')!='offsite_fetched_pending_isolated_restore'
+                or not a.source.name.endswith('.zip.age') or a.output.exists()
+                or not 0<a.source.stat().st_size<=20*1024**3
+                or source.get('ciphertext_sha256')!=digest(a.source)):
+            raise ValueError('verified_source_required')
+        k=['kubectl','--kubeconfig',a.kubeconfig];assert_cluster(k)
+        # Existing governed databases delivery, captured only in process memory.
+        r=subprocess.run(k+['-n','databases','get','secret','platform-pg-backup-s3','-o','json'],capture_output=True,timeout=30,check=True)
+        values=json.loads(r.stdout)['data']
+        import boto3
+        from botocore.config import Config
+        client=boto3.client('s3',endpoint_url=ENDPOINT,region_name='nl-ams',
+            aws_access_key_id=base64.b64decode(values['ACCESS_KEY_ID']).decode(),
+            aws_secret_access_key=base64.b64decode(values['ACCESS_SECRET_KEY']).decode(),
+            config=Config(request_checksum_calculation='when_required',response_checksum_validation='when_required',signature_version='s3v4',connect_timeout=15,read_timeout=180,retries={'max_attempts':3},s3={'addressing_style':'path'}))
+        transfer(client,a.source,a.output,receipt,checkpoint)
+    except Exception:
+        receipt.update(status='failed',error='bounded_primary_archive_transfer_failed');checkpoint()
+        return 1
+    return 0
+
+if __name__=='__main__': raise SystemExit(main())

tests/test_backup_tiers.py

@@ -0,0 +1,49 @@
+import io
+from pathlib import Path
+import sys
+import tempfile
+import unittest
+from unittest.mock import MagicMock,patch
+import zipfile
+sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts'))
+import scaleway_forgejo_archive as primary
+from capture_forgejo_archive import validate_archive
+
+class BackupTiers(unittest.TestCase):
+    def archive(self,path,package=False):
+        with zipfile.ZipFile(path,'w') as z:
+            z.writestr('forgejo-db.sql','fixture');z.writestr('repos/owner/repo.git/HEAD','ref: refs/heads/main')
+            z.writestr('data/lfs/unique','keep');z.writestr('data/attachments/unique','keep')
+            if package: z.writestr('data/packages/blob','bulk')
+    def test_full_keeps_packages_essentials_rejects_them(self):
+        with tempfile.TemporaryDirectory() as d:
+            p=Path(d)/'archive.zip';self.archive(p,True);validate_archive(p)
+            with self.assertRaises(ValueError): validate_archive(p,'essentials')
+    def test_essentials_keeps_unique_files(self):
+        with tempfile.TemporaryDirectory() as d:
+            p=Path(d)/'archive.zip';self.archive(p);validate_archive(p,'essentials')
+    def client(self):
+        c=MagicMock();c.create_multipart_upload.return_value={'UploadId':'fixture'}
+        c.upload_part.return_value={'ETag':'part-etag'}
+        c.complete_multipart_upload.return_value={'VersionId':'fixture-version'}
+        return c
+    def test_failed_upload_aborts_only_its_multipart(self):
+        with tempfile.TemporaryDirectory() as d:
+            p=Path(d)/'source';p.write_bytes(b'abc');c=self.client();c.upload_part.side_effect=RuntimeError('fixture')
+            receipt={}
+            with self.assertRaises(RuntimeError): primary.transfer(c,p,Path(d)/'out',receipt)
+            c.abort_multipart_upload.assert_called_once();c.delete_object.assert_not_called()
+            self.assertTrue(receipt['multipart_aborted'])
+    def test_download_is_version_pinned_and_hash_verified(self):
+        with tempfile.TemporaryDirectory() as d:
+            p=Path(d)/'source';p.write_bytes(b'abc');c=self.client()
+            c.get_object.return_value={'ContentLength':3,'Body':io.BytesIO(b'abc')}
+            receipt={};primary.transfer(c,p,Path(d)/'out',receipt)
+            self.assertEqual(c.get_object.call_args.kwargs['VersionId'],'fixture-version')
+            self.assertTrue(receipt['download_hash_matches']);c.abort_multipart_upload.assert_not_called()
+    def test_corrupt_download_does_not_pass(self):
+        with tempfile.TemporaryDirectory() as d:
+            p=Path(d)/'source';p.write_bytes(b'abc');c=self.client()
+            c.get_object.return_value={'ContentLength':3,'Body':io.BytesIO(b'xyz')}
+            with self.assertRaises(ValueError): primary.transfer(c,p,Path(d)/'out',{})
+            c.delete_object.assert_not_called()

workplans/RPF-WP-0038-forgejo-scaleway-primary-coverage.md

@@ -4,7 +4,7 @@ type: workplan
 title: "Close Forgejo primary backup coverage on Scaleway"
 domain: financials
 repo: railiance-platform
-status: blocked
+status: active
 owner: codex
 created: "2026-09-06"
 updated: "2026-09-06"
@@ -71,7 +71,7 @@ removed. Evidence: `docs/evidence/forgejo-scaleway-restore-2026-09-06.json`.
 
 ```task
 id: RPF-WP-0038-T04
-status: wait
+status: progress
 priority: high
 state_hub_task_id: "a4807df0-ec96-58f1-9cbd-42b1dcde0d6f"
 ```