diff --git a/infra/attic-migration/README.md b/infra/attic-migration/README.md index ee4958b3..ebf86252 100644 --- a/infra/attic-migration/README.md +++ b/infra/attic-migration/README.md @@ -4,6 +4,28 @@ Local-only operator tool for safe resumable Attic cache repack/migration. Parent ## Deployment layout +### Operational state — 2026-09-10 + +The primary `/hectic` endpoint now serves `nix-cache-hectic-lab` through +`atticd-repacked`; the cutover was applied with NixOS `switch`. The original +bucket/database remain preserved and readable at `/previous/hectic`. +`/next/hectic` is an alias for the new backend. Existing public keys and CI +tokens remain valid, and the primary endpoint is writable again. + +The migrated inventory contains 1343 paths and 1195 unique NAR hashes. The +independent inventory comparison and aggregate full-read receipts are recorded +in `/var/lib/attic-repack/verification-receipt.json`. Transient S3 504/read errors +required retries; this is data-integrity evidence, not a claim that Hetzner's +read availability is fixed. + +All migration, verification, seeding, and watcher jobs have been stopped for +user-controlled load testing. Do not automatically restart bulk verification. +The current generation is +`/nix/store/s7x1n9zprjzagb9pvkl0k4igdgnbbchh-nixos-system-hectic-lab-25.11.20260526.25f5383`. +The pinned rollback generation remains at +`/var/lib/attic-repack/rollback-system`; backups remain private under +`/var/lib/attic-repack/backups` and include the cache signing key. + - Original backend: `atticd`, port 8081, `/var/lib/atticd/server.db`, bucket `cache-hectic-lab` in HEL1. - During the write freeze and after cutover the original backend runs in @@ -30,9 +52,11 @@ Local-only operator tool for safe resumable Attic cache repack/migration. Parent Keep backups and manifests under private `/var/lib/attic-repack`; the SQLite backup includes the cache's private signing key. 4. Restart the original backend for reads only, refresh the complete inventory, - migrate any final delta, then run unfiltered `verify`. Its exit status must be - zero; independently compare old/new store-path, NAR hash, size and metadata - inventories. `status` alone is not a cutover certificate. + migrate any final delta, then run unfiltered `verify` across all 1343 old + paths. Its exit status must be zero with zero exhausted payload verification + failures; independently compare old/new store-path, NAR hash, size and + metadata inventories from the databases. `status` alone is not a cutover + certificate. 5. Pin the old/staging NixOS generation as a GC root, set `repackedActive = true`, build, inspect dry activation, and switch. `/hectic` now reaches the new backend; old data and `/previous/hectic` remain available. @@ -64,7 +88,7 @@ Default state dir: `/var/lib/attic-repack` (`0700`). Raw NAR spool path: /var/lib/attic-repack/raw/{sha256hex}.nar ``` -Parent may seed this file directly. Tool always verifies SHA-256 and byte length before upload. Checkpoints live under `checkpoints/{store_path_hash}.json` and contain no keypair/token. +Parent may seed this file directly. Tool always verifies SHA-256 and byte length before upload. Checkpoints live under `checkpoints/{store_path_hash}.json` and contain no keypair/token. Forced payload verification records a receipt with `payload_verified_at`, `payload_verify_attempts`, `payload_sha256`, and `payload_bytes` only after a complete successful read. ## Commands @@ -109,12 +133,23 @@ Records also include upload metadata: `store_path_hash`, `references`, `system`, After stopping old writers and taking a consistent snapshot, run an unfiltered `verify` (no `--paths-file` or `--limit`) to reread every new NAR and reconcile all paths, metadata, hashes, and sizes before switching the primary endpoint. + Do not skip files, change expected hashes, or relax server/client timeouts to + pass this gate. - A local store path can differ from the historical cached NAR. Such a local copy is rejected and recovered from the original S3 chunks instead. - Old DB is opened readonly; old SQL NAR/chunk tables are never copied. - Missing local raw NARs are reconstructed from old S3 chunkrefs with per-object retries and chunk/full hash checks. - Upload uses Attic `PUT /_api/v1/upload-path` with JSON preamble plus raw uncompressed NAR. -- New cache verification compares immutable metadata against old rendered narinfo and reads/decompresses one payload per verified path invocation. +- New cache verification compares immutable metadata against old rendered narinfo + and reads/decompresses one payload per verified path invocation. Payload reads + make up to three fresh attempts for transport HTTP 408/429/5xx and truncated + body/decompressor EOF failures only. Each attempt follows a new GET/redirect, + starts SHA-256 and byte counts from zero, closes failed readers, and fails + immediately on hash mismatch, full-size mismatch, oversized payload, missing + URL, unsupported compression, or HTTP 4xx other than 408/429. +- A receipt with retries proves the path was fully read and matched integrity; it + does not prove the storage provider is healthy. Treat retry events as provider + health signals separate from cutover correctness. - Authenticated HTTP is refused unless URL host is loopback. ## Local build/test diff --git a/infra/attic-migration/repack.py b/infra/attic-migration/repack.py index 8c6ab9f5..3171ef97 100644 --- a/infra/attic-migration/repack.py +++ b/infra/attic-migration/repack.py @@ -14,8 +14,10 @@ import argparse import concurrent.futures from collections import deque import contextlib +import datetime import gzip import hashlib +import http.client import itertools import io import json @@ -51,6 +53,14 @@ class RepackError(RuntimeError): """Expected operational failure with sanitized message.""" +class PayloadIntegrityError(RepackError): + """Verified payload differs from immutable expected NAR identity.""" + + +class PayloadTransientError(RepackError): + """Retryable transport or truncated payload read failure.""" + + def eprint(*args: object) -> None: print(*args, file=sys.stderr, flush=True) @@ -59,6 +69,10 @@ def now() -> float: return time.time() +def utc_timestamp() -> str: + return datetime.datetime.now(datetime.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + def require_private(path: pathlib.Path, directory: bool) -> None: mode = 0o700 if directory else 0o600 if directory: @@ -257,6 +271,40 @@ def sanitized_error(exc: BaseException) -> str: return exc.__class__.__name__ +def http_status_from_error(exc: BaseException) -> int | None: + if not isinstance(exc, RepackError): + return None + parts = str(exc).split() + if len(parts) >= 2 and parts[0] == "HTTP": + with contextlib.suppress(ValueError): + return int(parts[1]) + return None + + +def payload_error_class(exc: BaseException) -> str: + status = http_status_from_error(exc) + if status is not None: + return f"HTTP{status}" + return exc.__class__.__name__ + + +def is_transient_payload_error(exc: BaseException) -> bool: + if isinstance(exc, PayloadIntegrityError): + return False + if isinstance(exc, PayloadTransientError): + return True + status = http_status_from_error(exc) + if status is not None: + return status in (408, 429) or 500 <= status <= 599 + if isinstance(exc, (TimeoutError, EOFError, http.client.IncompleteRead, http.client.HTTPException)): + return True + module = exc.__class__.__module__.split(".", 1)[0] + name = exc.__class__.__name__ + if module in ("requests", "urllib3"): + return name in {"ConnectionError", "ReadTimeout", "Timeout", "ProtocolError", "ChunkedEncodingError"} + return name in {"ProtocolError", "IncompleteRead", "BadGzipFile", "LZMAError", "ZstdError"} + + def safe_headers(host: str | None = None, token: str | None = None) -> dict[str, str]: headers: dict[str, str] = {} if host: @@ -395,6 +443,8 @@ class AtticClient: current = next_url continue if resp.status_code >= 400: + with contextlib.suppress(Exception): + resp.close() raise RepackError(f"HTTP {resp.status_code} GET {urllib.parse.urlsplit(current).path}") return resp raise RepackError("too many public redirects") @@ -452,13 +502,37 @@ class AtticClient: raise time.sleep(0.5 * (2**attempt)) - def verify_payload(self, narinfo: dict[str, Any], expected_hash: str, expected_size: int, narinfo_url: str) -> None: + def verify_payload(self, narinfo: dict[str, Any], expected_hash: str, expected_size: int, narinfo_url: str, store_path: str | None = None, max_attempts: int = 3) -> dict[str, Any]: raw_url = narinfo.get("URL") if not raw_url: raise RepackError("new narinfo missing URL") url = urllib.parse.urljoin(narinfo_url, raw_url) - resp = self._public_stream(url) compression = (narinfo.get("Compression") or pathlib.PurePosixPath(raw_url).suffix.lstrip(".")).lower() + if compression not in ("zstd", "zst", "", "none", "gzip", "gz", "xz"): + raise RepackError(f"unsupported new nar compression {compression}") + + for attempt in range(1, max_attempts + 1): + try: + digest, size = self._verify_payload_attempt(url, compression, expected_hash, expected_size) + return {"attempts": attempt, "sha256": digest, "bytes": size} + except Exception as exc: + if not is_transient_payload_error(exc) or attempt >= max_attempts: + raise + event = { + "payload_verify_retry": True, + "publicStorePath": store_path, + "expectedNARhash": "sha256:" + expected_hash, + "expectedNARsize": expected_size, + "attempt": attempt, + "max_attempts": max_attempts, + "errorclass": payload_error_class(exc), + } + eprint(json.dumps(event, sort_keys=True)) + time.sleep(0.5 * (2 ** (attempt - 1))) + raise RepackError("payload verification retry loop exhausted") + + def _verify_payload_attempt(self, url: str, compression: str, expected_hash: str, expected_size: int) -> tuple[str, int]: + resp = self._public_stream(url) h = hashlib.sha256() size = 0 source = resp.raw @@ -475,16 +549,27 @@ class AtticClient: try: with contextlib.closing(reader): while True: - data = reader.read(CHUNK) + try: + data = reader.read(CHUNK) + except Exception as exc: + if is_transient_payload_error(exc): + raise PayloadTransientError(payload_error_class(exc)) from exc + raise if not data: break h.update(data) size += len(data) + if size > expected_size: + raise PayloadIntegrityError("new NAR payload larger than expected") finally: with contextlib.suppress(Exception): resp.close() - if h.hexdigest() != expected_hash or size != expected_size: - raise RepackError("new NAR payload hash/size mismatch") + digest = h.hexdigest() + if size < expected_size: + raise PayloadTransientError("short new NAR payload") + if digest != expected_hash or size != expected_size: + raise PayloadIntegrityError("new NAR payload hash/size mismatch") + return digest, size class PrefixFileBody: @@ -539,7 +624,7 @@ class InventoryDB: return con def cache_row(self) -> dict[str, Any]: - with self.connect() as con: + with contextlib.closing(self.connect()) as con: row = con.execute( "select name,keypair,is_public,store_dir,priority,upstream_cache_key_names,retention_period " "from cache where name=? and deleted_at is null", @@ -558,7 +643,7 @@ class InventoryDB: "order by o.store_path" ) args: list[Any] = [self.cache] - with self.connect() as con: + with contextlib.closing(self.connect()) as con: rows = con.execute(sql, args).fetchall() out: list[dict[str, Any]] = [] for row in rows: @@ -580,7 +665,7 @@ class InventoryDB: "select cr.seq,ch.chunk_hash,ch.chunk_size,ch.file_hash,ch.file_size,ch.compression,ch.remote_file,ch.state " "from chunkref cr join chunk ch on ch.id=cr.chunk_id where cr.nar_id=? order by cr.seq" ) - with self.connect() as con: + with contextlib.closing(self.connect()) as con: return [dict(r) for r in con.execute(sql, (nar_id,)).fetchall()] @@ -620,7 +705,7 @@ class State: def get_checkpoint(self, record: dict[str, Any]) -> dict[str, Any]: return load_json(self.checkpoint_path(record["store_path_hash"]), {}) - def set_checkpoint(self, record: dict[str, Any], status: str, tries: int, error: str | None = None) -> None: + def set_checkpoint(self, record: dict[str, Any], status: str, tries: int, error: str | None = None, payload_receipt: dict[str, Any] | None = None) -> None: value = { "store_path": record["store_path"], "store_path_hash": record["store_path_hash"], @@ -632,6 +717,19 @@ class State: } if error: value["error"] = sanitized_error(RepackError(error)) + if status == "verified": + if payload_receipt is not None: + value.update({ + "payload_verified_at": utc_timestamp(), + "payload_verify_attempts": int(payload_receipt["attempts"]), + "payload_sha256": str(payload_receipt["sha256"]), + "payload_bytes": int(payload_receipt["bytes"]), + }) + else: + previous = self.get_checkpoint(record) + for key in ("payload_verified_at", "payload_verify_attempts", "payload_sha256", "payload_bytes"): + if key in previous: + value[key] = previous[key] atomic_json(self.checkpoint_path(record["store_path_hash"]), value) @@ -870,11 +968,13 @@ class Migrator: tries = int(self.state.get_checkpoint(record).get("tries", 0)) + 1 try: if verify_only: - if not self.verify_record(record, force_payload=True): + result = self.verify_record(record, force_payload=True) + if not result: raise RepackError("new narinfo missing") else: - self.migrate_record(record) - self.state.set_checkpoint(record, "verified", tries) + result = self.migrate_record(record) + receipt = result if isinstance(result, dict) else None + self.state.set_checkpoint(record, "verified", tries, payload_receipt=receipt) q.put(("ok", record["store_path"])) except Exception as exc: self.state.set_checkpoint(record, "failed", tries, sanitized_error(exc)) @@ -893,20 +993,22 @@ class Migrator: fut.result() return failures - def migrate_record(self, record: dict[str, Any]) -> None: + def migrate_record(self, record: dict[str, Any]) -> dict[str, Any] | None: cp = self.state.get_checkpoint(record) if cp.get("status") == "verified" and cp.get("metadata_fingerprint") == metadata_fingerprint(record): if self.verify_record(record, force_payload=False): - return + return None nar_path = self.ensure_raw_nar(record) actual_hash, actual_size = sha256_file(nar_path) if actual_hash != nar_hash_hex(record["nar_hash"]) or actual_size != int(record["nar_size"]): raise RepackError("raw NAR spool hash/size mismatch") self.new_client.upload(record, nar_path) - if not self.verify_record(record, force_payload=True): + result = self.verify_record(record, force_payload=True) + if not result: raise RepackError("new narinfo missing after upload") + return result if isinstance(result, dict) else None - def verify_record(self, record: dict[str, Any], force_payload: bool) -> bool: + def verify_record(self, record: dict[str, Any], force_payload: bool) -> bool | dict[str, Any]: narinfo = self.new_client.get_narinfo(record["store_path_hash"]) if narinfo is None: return False @@ -921,7 +1023,7 @@ class Migrator: if diffs: raise RepackError("new narinfo immutable metadata mismatch: " + ",".join(diffs)) if force_payload: - self.new_client.verify_payload(narinfo, nar_hash_hex(record["nar_hash"]), int(record["nar_size"]), self.new_client.narinfo_url(record["store_path_hash"])) + return self.new_client.verify_payload(narinfo, nar_hash_hex(record["nar_hash"]), int(record["nar_size"]), self.new_client.narinfo_url(record["store_path_hash"]), record.get("store_path")) return True def ensure_raw_nar(self, record: dict[str, Any]) -> pathlib.Path: diff --git a/infra/attic-migration/test_repack.py b/infra/attic-migration/test_repack.py index 5b4439db..e68c868e 100644 --- a/infra/attic-migration/test_repack.py +++ b/infra/attic-migration/test_repack.py @@ -26,12 +26,13 @@ class FakeResponse: self._json = json_data self.raw = raw or io.BytesIO(content) self.headers = headers or {} + self.close_count = 0 def json(self): return self._json def close(self): - pass + self.close_count += 1 class FakeSession: @@ -279,6 +280,61 @@ class RepackTests(unittest.TestCase): self.assertNotEqual(seen["s3_host"], "cache.hectic-lab.com") self.assertIsNone(seen["s3_auth"]) + def test_payload_retry_truncated_then_full_resets_hash(self): + data = b"complete NAR bytes" + client = repack.AtticClient("http://127.0.0.1:8082", "hectic", None, None) + sess = FakeSession(); client._local.session = sess + responses = [FakeResponse(200, content=data[:4]), FakeResponse(200, content=data)] + + def get(_method, _url, _kwargs): + return responses.pop(0) + + sess.routes[("GET", "http://127.0.0.1:8082/hectic/nar/x")] = get + with mock.patch("time.sleep") as sleep: + receipt = client.verify_payload({"URL": "nar/x", "Compression": "none"}, sha(data), len(data), "http://127.0.0.1:8082/hectic/abcd.narinfo", "/nix/store/abcd-name") + self.assertEqual(receipt, {"attempts": 2, "sha256": sha(data), "bytes": len(data)}) + self.assertEqual(len(sess.calls), 2) + self.assertEqual(sleep.call_count, 1) + + def test_payload_persistent_timeouts_fail_after_three(self): + client = repack.AtticClient("http://127.0.0.1:8082", "hectic", None, None) + sess = FakeSession(); client._local.session = sess + + def timeout(_method, _url, _kwargs): + raise TimeoutError() + + sess.routes[("GET", "http://127.0.0.1:8082/hectic/nar/x")] = timeout + with mock.patch("time.sleep") as sleep, self.assertRaises(TimeoutError): + client.verify_payload({"URL": "nar/x", "Compression": "none"}, sha(b"x"), 1, "http://127.0.0.1:8082/hectic/abcd.narinfo") + self.assertEqual(len(sess.calls), 3) + self.assertEqual(sleep.call_count, 2) + + def test_payload_full_size_wrong_hash_fails_after_one(self): + client = repack.AtticClient("http://127.0.0.1:8082", "hectic", None, None) + sess = FakeSession(); client._local.session = sess + sess.routes[("GET", "http://127.0.0.1:8082/hectic/nar/x")] = FakeResponse(200, content=b"bad") + with self.assertRaises(repack.PayloadIntegrityError): + client.verify_payload({"URL": "nar/x", "Compression": "none"}, sha(b"nar"), 3, "http://127.0.0.1:8082/hectic/abcd.narinfo") + self.assertEqual(len(sess.calls), 1) + + def test_payload_oversize_fails_after_one(self): + client = repack.AtticClient("http://127.0.0.1:8082", "hectic", None, None) + sess = FakeSession(); client._local.session = sess + sess.routes[("GET", "http://127.0.0.1:8082/hectic/nar/x")] = FakeResponse(200, content=b"toolong") + with self.assertRaises(repack.PayloadIntegrityError): + client.verify_payload({"URL": "nar/x", "Compression": "none"}, sha(b"too"), 3, "http://127.0.0.1:8082/hectic/abcd.narinfo") + self.assertEqual(len(sess.calls), 1) + + def test_payload_http_403_no_retry_and_closes(self): + client = repack.AtticClient("http://127.0.0.1:8082", "hectic", None, None) + sess = FakeSession(); client._local.session = sess + resp = FakeResponse(403) + sess.routes[("GET", "http://127.0.0.1:8082/hectic/nar/x")] = resp + with self.assertRaises(repack.RepackError): + client.verify_payload({"URL": "nar/x", "Compression": "none"}, sha(b"x"), 1, "http://127.0.0.1:8082/hectic/abcd.narinfo") + self.assertEqual(len(sess.calls), 1) + self.assertEqual(resp.close_count, 1) + def test_auth_api_redirect_refused(self): tp = repack.TokenProvider(None, None, "hectic") tp._token = "tok"; tp._expires = repack.now() + 3600 @@ -409,7 +465,7 @@ class RepackTests(unittest.TestCase): mig.selected_records = lambda: [record] mig.new_client = mock.Mock() mig.new_client.get_narinfo.return_value = {**repack.expected_narinfo(record), "URL": "nar/x", "Compression": "none"} - mig.new_client.verify_payload.return_value = None + mig.new_client.verify_payload.return_value = {"attempts": 1, "sha256": sha(b"abc"), "bytes": 3} mig.new_client.narinfo_url.return_value = "http://127.0.0.1:8082/hectic/abcd.narinfo" mig.old_client = mock.Mock() mig.old_client.get_narinfo.return_value = {**repack.expected_narinfo(record)} @@ -417,6 +473,38 @@ class RepackTests(unittest.TestCase): mig.new_client.upload.assert_not_called() mig.new_client.verify_payload.assert_called_once() + def test_payload_receipt_only_for_forced_successful_full_read(self): + with tempfile.TemporaryDirectory() as td: + mig = repack.Migrator(self.args(td)) + record = self.record(b"abc") + mig.selected_records = lambda: [record] + narinfo = {**repack.expected_narinfo(record), "URL": "nar/x", "Compression": "none"} + old_info = {**repack.expected_narinfo(record)} + mig.old_client = mock.Mock(); mig.old_client.get_narinfo.return_value = old_info + mig.new_client = mock.Mock(); mig.new_client.get_narinfo.return_value = narinfo; mig.new_client.narinfo_url.return_value = "http://127.0.0.1/hectic/abcd.narinfo" + mig.new_client.verify_payload.return_value = {"attempts": 2, "sha256": sha(b"abc"), "bytes": 3} + self.assertEqual(mig.migrate(True), 0) + cp = mig.state.get_checkpoint(record) + self.assertEqual(cp["payload_verify_attempts"], 2) + self.assertEqual(cp["payload_sha256"], sha(b"abc")) + self.assertEqual(cp["payload_bytes"], 3) + self.assertTrue(cp["payload_verified_at"].endswith("Z")) + first_verified_at = cp["payload_verified_at"] + + mig.new_client.verify_payload.reset_mock() + self.assertEqual(mig.migrate(False), 0) + cp = mig.state.get_checkpoint(record) + self.assertEqual(cp["payload_verified_at"], first_verified_at) + mig.new_client.verify_payload.assert_not_called() + + mig.new_client.get_narinfo.return_value = narinfo + mig.new_client.verify_payload.side_effect = repack.PayloadIntegrityError("new NAR payload hash/size mismatch") + self.assertEqual(mig.migrate(True), 1) + cp = mig.state.get_checkpoint(record) + self.assertEqual(cp["status"], "failed") + self.assertNotIn("payload_verified_at", cp) + self.assertNotIn("payload_sha256", cp) + def test_key_redaction(self): secret = "eyJhbGciOiPRIVATEKEYX-Amz-Signature=abc" msg = repack.sanitized_error(RuntimeError("https://x/y?" + secret)) diff --git a/nixos/system/hectic-lab/attic.nix b/nixos/system/hectic-lab/attic.nix index 5fe8cd59..483e16ce 100644 --- a/nixos/system/hectic-lab/attic.nix +++ b/nixos/system/hectic-lab/attic.nix @@ -7,8 +7,8 @@ pkgs, ... }: let - repackedActive = false; - migrationWriteFreeze = true; + repackedActive = true; + migrationWriteFreeze = false; repackedSettings = config.services.atticd.settings // { listen = "127.0.0.1:8082";