优化(08-28-production-arm-backup-transfer-incident): 收窄生产哈希读取

.trellis/spec/hyb-oa/deployment-trialrun.md

@@ -2872,6 +2872,12 @@ current shared mapping -> fine-grained stages + mutation listeners
   bytes and one of `production_single_point` or `no_proven_recovery_point`.
   Off-host staleness alone does not distinguish those two materially different
   production risks.
+- On an online production host, enumerate older local batches with manifest,
+  file-set, identity and byte metadata only. Fully hash exactly the newest
+  metadata-valid candidate; report older hashes as `NOT_RUN`. If that latest
+  candidate fails SHA-256, keep `no_proven_recovery_point` and do not fall back
+  to multi-GiB reads of older candidates during the same probe. ARM archive
+  validation remains full-contract because it is the backup node.
 - Inspect cleanup semantics as executable ordering, not a remembered claim.
   Source publication must precede transfer; local pruning must follow verified
   off-host finalize and success-state publication; generation staging must be
@@ -2959,6 +2965,10 @@ current shared mapping -> fine-grained stages + mutation listeners
   corruption and malformed manifest. Assert the risk classification and
   transition recommendation change without emitting paths, run ids, artifact
   names, raw diagnostics or tracebacks.
+- Fixture two metadata-valid production batches with a same-size hash mutation
+  in the older and then newest candidate. Assert only the newest candidate is
+  fully hashed, older hashes remain `NOT_RUN`, and a newest-candidate hash
+  failure does not trigger fallback reads.
 - Mechanically mutate the wrapper ordering so pruning appears before transfer;
   the cleanup/retention contract must turn false. A prose claim that failure
   retains data is not sufficient regression evidence.

.trellis/tasks/08-28-production-arm-backup-transfer-incident/design.md

@@ -31,7 +31,7 @@
 
 ## 只读采集合同
 
-- 生产侧首先输出最新本地候选是否通过完整 manifest/三文件/字节/SHA-256 合同、年龄、完整批次数量和聚合大小；不输出 run id、文件名或路径。该结果先于传输根因分析，决定当前是“生产单点”还是“无已证明可用备份”。
+- 生产侧先用 manifest/三文件/身份/字节元数据枚举全部本地候选，只对最新一个元数据合格候选完整读取 database/storage 并计算 SHA-256；旧候选明确输出 `older_metadata_batch_hash_status=NOT_RUN`，不得为了计数连续读取数 GiB。最新候选只有通过完整 SHA-256 合同才输出 `complete_local_recovery_point=yes`；其哈希失败时不得现场回退扫描旧候选，而应保守标为“无已证明可用备份”。输出年龄、元数据合格数量和聚合大小，不输出 run id、文件名或路径。该结果先于传输根因分析，决定当前是“生产单点”还是“无已证明可用备份”。ARM 仍对正式归档执行完整哈希合同。
 - 对成功与失败路径分别输出本地完整批次、生成期临时对象和保留策略的聚合结果，明确 `retained_after_failure`、`pruned_after_success` 和连续失败累积趋势；不从单次目录计数猜测脚本语义。
 - partial 只输出每个匿名样本的总字节数、预期总字节数、安全比例和年龄；不输出名称、路径、run id 或 artifact 名称。
 - 生产私有诊断只输出固定的 `transfer_exit_code`、`failure_category`、开始/结束 epoch 和持续秒数。若现有状态没有持久化真实 rsync 码或开始时刻，记录 `NOT_AVAILABLE`，需要本地设计可观测性修订并另行批准部署，不能现场改脚本。

.trellis/tasks/08-28-production-arm-backup-transfer-incident/implement.md

@@ -16,7 +16,7 @@
 ### 本地准备结果（2026-08-28）
 
 - 跟踪调查器：`scripts/qa/inspect-backup-transfer-incident.py`；仅接受显式绝对路径参数，不回显参数值，未知异常也只输出稳定类别。
-- 可复跑合同：`scripts/qa/run-backup-transfer-incident-contract.sh`；17 项合成检查通过，fixture 与 Python cache 零残留；权限拒绝不会被误写成“无有效备份”，其他阶段的旧日志不会冒充本轮 rsync 诊断。
+- 可复跑合同：`scripts/qa/run-backup-transfer-incident-contract.sh`；19 项合成检查通过，fixture 与 Python cache 零残留；权限拒绝不会被误写成“无有效备份”，其他阶段的旧日志不会冒充本轮 rsync 诊断；生产多个批次时只哈希最新元数据合格候选，最新哈希失败不回退扫描旧批次。
 - 既有 `run-automatic-backup-contract.sh` 74 项通过；生产 wrapper 未改。现状没有持久化传输专属开始/结束时刻，故持续时间必须为 `NOT_AVAILABLE`。
 - rsync 原始码只在安全摘要明确为 `phase_offhost_transfer_failed` 时从 root 私有 `last-error.log` 固定正则提取，避免把 paired-backup/finalize 的旧诊断错误归因给传输。
 

scripts/qa/inspect-backup-transfer-incident.py

@@ -33,6 +33,7 @@ SIMILAR_RELATIVE_PPM = 50_000
 SIMILAR_ABSOLUTE_BYTES = 64 * 1024 * 1024
 SAFE_KEY_RE = re.compile(r"^[a-z0-9_]+$")
 SAFE_VALUE_RE = re.compile(r"^(?:[A-Za-z][A-Za-z0-9_]*|-?[0-9]+)$")
+SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
 
 
 class InspectionError(Exception):
@@ -115,7 +116,13 @@ def run_epoch(stamp: str) -> int | None:
     return int(parsed.timestamp())
 
 
-def validate_batch(path: Path, expected_run_id: str, stamp: str) -> BatchResult:
+def validate_batch(
+    path: Path,
+    expected_run_id: str,
+    stamp: str,
+    *,
+    verify_hashes: bool = True,
+) -> BatchResult:
     total = tree_bytes(path)
     received_artifact_bytes = 0
     expected_artifact_bytes: int | None = None
@@ -149,10 +156,15 @@ def validate_batch(path: Path, expected_run_id: str, stamp: str) -> BatchResult:
             expected_size = item.get("bytes")
             if not isinstance(expected_size, int) or expected_size < 0:
                 return BatchResult(False, total, created_epoch, None, 0)
+            expected_sha = item.get("sha256")
+            if not isinstance(expected_sha, str) or not SHA256_RE.fullmatch(expected_sha):
+                return BatchResult(False, total, created_epoch, None, 0)
             expected_artifact_bytes += expected_size
             actual_size = target.stat().st_size
             received_artifact_bytes += actual_size
-            if actual_size != expected_size or file_sha256(target) != item.get("sha256"):
+            if actual_size != expected_size or (
+                verify_hashes and file_sha256(target) != expected_sha
+            ):
                 return BatchResult(
                     False,
                     total,
@@ -329,14 +341,19 @@ def production(args: argparse.Namespace) -> None:
     backup_root = safe_root(args.backup_root)
     state_dir = safe_root(args.state_dir)
     now = args.now_epoch
-    valid: list[BatchResult] = []
+    metadata_valid: list[tuple[Path, str, str, BatchResult]] = []
     invalid_count = invalid_bytes = temp_count = temp_bytes = other_count = 0
     for entry in backup_root.iterdir():
         match = BATCH_RE.fullmatch(entry.name)
         if match and entry.is_dir() and not entry.is_symlink():
-            result = validate_batch(entry, match.group(1), match.group(2))
+            result = validate_batch(
+                entry,
+                match.group(1),
+                match.group(2),
+                verify_hashes=False,
+            )
             if result.valid and result.created_epoch is not None and result.created_epoch <= now:
-                valid.append(result)
+                metadata_valid.append((entry, match.group(1), match.group(2), result))
             else:
                 invalid_count += 1
                 invalid_bytes += result.total_bytes
@@ -346,27 +363,62 @@ def production(args: argparse.Namespace) -> None:
         else:
             other_count += 1
 
-    valid.sort(key=lambda item: item.created_epoch or -1)
+    metadata_valid.sort(key=lambda item: item[3].created_epoch or -1)
-    latest = valid[-1] if valid else None
+    latest_metadata = metadata_valid[-1] if metadata_valid else None
+    latest_verified: BatchResult | None = None
+    if latest_metadata is not None:
+        latest_verified = validate_batch(
+            latest_metadata[0],
+            latest_metadata[1],
+            latest_metadata[2],
+            verify_hashes=True,
+        )
+    complete_local_recovery_point = bool(latest_verified and latest_verified.valid)
+    latest_result = latest_metadata[3] if latest_metadata else None
     output: list[tuple[str, object]] = [
         ("production_probe_result", "PASS"),
-        ("complete_local_recovery_point", bool(valid)),
+        ("complete_local_recovery_point", complete_local_recovery_point),
-        ("backup_risk_classification", "production_single_point" if valid else "no_proven_recovery_point"),
-        ("complete_local_batch_count", len(valid)),
-        ("complete_local_batch_total_bytes", sum(item.total_bytes for item in valid)),
         (
-            "latest_complete_local_batch_age_seconds",
+            "backup_risk_classification",
-            now - latest.created_epoch
+            "production_single_point"
-            if latest and latest.created_epoch is not None
+            if complete_local_recovery_point
+            else "no_proven_recovery_point",
+        ),
+        ("fully_verified_local_batch_count", 1 if complete_local_recovery_point else 0),
+        ("metadata_complete_local_batch_count", len(metadata_valid)),
+        (
+            "metadata_complete_local_batch_total_bytes",
+            sum(item[3].total_bytes for item in metadata_valid),
+        ),
+        (
+            "older_metadata_batch_hash_status",
+            "NOT_RUN" if len(metadata_valid) > 1 else "NOT_APPLICABLE",
+        ),
+        (
+            "latest_local_candidate_hash_status",
+            "PASS"
+            if complete_local_recovery_point
+            else ("FAIL" if latest_metadata is not None else "NOT_RUN"),
+        ),
+        (
+            "latest_local_candidate_age_seconds",
+            now - latest_result.created_epoch
+            if latest_result and latest_result.created_epoch is not None
             else "NOT_AVAILABLE",
         ),
-        ("latest_complete_local_batch_bytes", latest.total_bytes if latest else "NOT_AVAILABLE"),
+        (
+            "latest_local_candidate_bytes",
+            latest_result.total_bytes if latest_result else "NOT_AVAILABLE",
+        ),
         ("invalid_local_batch_count", invalid_count),
         ("invalid_local_batch_total_bytes", invalid_bytes),
         ("generation_temp_count", temp_count),
         ("generation_temp_total_bytes", temp_bytes),
         ("unclassified_local_entry_count", other_count),
-        ("transition_recovery_point_action", "wait_natural" if valid else "request_manual_run"),
+        (
+            "transition_recovery_point_action",
+            "wait_natural" if complete_local_recovery_point else "request_manual_run",
+        ),
     ]
     output.extend(filesystem_fields("production_backup", backup_root))
     output.extend(source_contract(args.wrapper_script, args.backup_script))

scripts/qa/run-backup-transfer-incident-contract.sh

@@ -95,6 +95,32 @@ assert_line "$TEST_ROOT/production-valid.out" "transfer_failure_category=partial
 assert_line "$TEST_ROOT/production-valid.out" "transfer_duration_seconds=NOT_AVAILABLE"
 pass "生产存在完整本地恢复点时识别为单点风险并保留未测时长"
 
+PRODUCTION_MULTIPLE="$TEST_ROOT/production-multiple"
+make_production_layout "$PRODUCTION_MULTIPLE"
+make_batch "$PRODUCTION_MULTIPLE/backups/hyb-oa-backup-auto-20260827T000000Z" "auto-20260827T000000Z"
+make_batch "$PRODUCTION_MULTIPLE/backups/hyb-oa-backup-auto-20260828T000000Z" "auto-20260828T000000Z"
+printf 'X' | dd of="$PRODUCTION_MULTIPLE/backups/hyb-oa-backup-auto-20260827T000000Z/database.dump" bs=1 seek=0 conv=notrunc status=none
+run_production "$PRODUCTION_MULTIPLE" "$TEST_ROOT/production-multiple.out"
+assert_line "$TEST_ROOT/production-multiple.out" "complete_local_recovery_point=yes"
+assert_line "$TEST_ROOT/production-multiple.out" "fully_verified_local_batch_count=1"
+assert_line "$TEST_ROOT/production-multiple.out" "metadata_complete_local_batch_count=2"
+assert_line "$TEST_ROOT/production-multiple.out" "older_metadata_batch_hash_status=NOT_RUN"
+assert_line "$TEST_ROOT/production-multiple.out" "latest_local_candidate_hash_status=PASS"
+pass "生产多个批次时只对最新元数据合格候选执行全量 SHA-256"
+
+PRODUCTION_LATEST_HASH_FAIL="$TEST_ROOT/production-latest-hash-fail"
+make_production_layout "$PRODUCTION_LATEST_HASH_FAIL"
+make_batch "$PRODUCTION_LATEST_HASH_FAIL/backups/hyb-oa-backup-auto-20260827T000000Z" "auto-20260827T000000Z"
+make_batch "$PRODUCTION_LATEST_HASH_FAIL/backups/hyb-oa-backup-auto-20260828T000000Z" "auto-20260828T000000Z"
+printf 'X' | dd of="$PRODUCTION_LATEST_HASH_FAIL/backups/hyb-oa-backup-auto-20260828T000000Z/database.dump" bs=1 seek=0 conv=notrunc status=none
+run_production "$PRODUCTION_LATEST_HASH_FAIL" "$TEST_ROOT/production-latest-hash-fail.out"
+assert_line "$TEST_ROOT/production-latest-hash-fail.out" "complete_local_recovery_point=no"
+assert_line "$TEST_ROOT/production-latest-hash-fail.out" "fully_verified_local_batch_count=0"
+assert_line "$TEST_ROOT/production-latest-hash-fail.out" "metadata_complete_local_batch_count=2"
+assert_line "$TEST_ROOT/production-latest-hash-fail.out" "older_metadata_batch_hash_status=NOT_RUN"
+assert_line "$TEST_ROOT/production-latest-hash-fail.out" "latest_local_candidate_hash_status=FAIL"
+pass "最新候选哈希失败时不回退扫描旧批次并保持无已证明恢复点"
+
 PRODUCTION_EMPTY="$TEST_ROOT/production-empty"
 make_production_layout "$PRODUCTION_EMPTY"
 run_production "$PRODUCTION_EMPTY" "$TEST_ROOT/production-empty.out"