diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bc77bc3b..a0b4460d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -25,6 +25,17 @@ jobs: - name: Test demo decisions without external services run: python3 -m unittest discover -s scripts/demo -p 'test_*.py' -v + recovery-contract: + runs-on: ubuntu-latest + timeout-minutes: 3 + steps: + - name: Check out repository + uses: actions/checkout@v7 + with: + persist-credentials: false + - name: Test isolated recovery decisions without cloud access + run: python3 -m unittest discover -s scripts/opensql -p 'test_dr_recovery.py' -v + backend: runs-on: ubuntu-latest timeout-minutes: 15 diff --git a/docs/design/issue-560-upload-recovery-contract.md b/docs/design/issue-560-upload-recovery-contract.md new file mode 100644 index 00000000..9d656d19 --- /dev/null +++ b/docs/design/issue-560-upload-recovery-contract.md @@ -0,0 +1,102 @@ +# 업로드 성공 응답의 DB·GCS 복원 판정 계약 + +## 목적과 범위 + +업로드 201 응답을 받았다는 사실과, 복원 DB·원본에서 해당 업로드가 보존됐다는 사실을 분리한다. 기존 Barman/WAL·GCS 보존 도구를 다시 만들지 않고, 클라이언트 응답 원장과 단계별 관문을 추가한다. 관련 이슈는 #560이다. + +이 변경은 로컬 Python 계약과 실행 제어기다. 실제 OpenSQL 복원, 실제 GCS 다운로드, BGE-M3 검색의 성공 증거는 후속 격리 실행에서 수집해야 한다. `PASS`는 입력한 연산자의 계약 충족을 뜻하며 RPO 0이나 공개 서비스 RTO를 보장하지 않는다. + +## 재사용할 기존 도구 + +| 구성요소 | 재사용 범위 | 추가로 필요한 관측 | +|---|---|---| +| `barman_wal_follow_primary.py` | Primary 대상 선택과 수신 재연결 | 실제 수신 프로세스·새 WAL 진행; dry-run만으로 정상 수신 판정 금지 | +| `barman_offsite_sync.py` | DONE 기준 백업과 완결 WAL 외부 전송 | 복원 가능한 WAL 연결·외부 identity; 마지막 전송 성공만으로 판정 금지 | +| `verify_gcs_original_preservation.py` | 장애 전 객체 복사·generation·해시 원장 | 장애 뒤에는 원본 접근 없이 고정 보존본만 읽어야 함 | +| `ha_evidence.py` | 0600 쓰기·fsync·원자적 파일 교체 | 이번 실험 전용 요청 ID·응답 ID·최초 감사 스키마 | +| `ha_worker_documents.py` / `analyze_worker_recovery.py` | 합성 문서와 최종 Job·Chunk·Embedding 대조 | 업로드 응답 ID 4종과 최초 복원 상태를 별도 보존 | + +```text +그림 1 — 장애 이전 응답 원장과 고정 입력 [실행 설계] + +시험 클라이언트 → record_sent(request_id, SHA-256, 크기, cohort) + → 실제 업로드 API 호출 + → record_response(HTTP 상태, 문서·버전·파일·Job ID) + ├─ 200/201 + ID 4종: ACKNOWLEDGED + ├─ 응답 미수신: UNKNOWN + └─ 다른 HTTP 상태: HTTP_ERROR (DB 롤백을 뜻하지 않음) + +백업 연산자 → 기준 백업·완결 WAL·원본 보존본 준비 + → identity 해시·객체 generation·SHA-256 고정 +시험 제어기 → validate_inputs: 준비 조건 7종 모두 참인지 확인 + → freeze_inputs: inputs.json 단독 생성 + cutoff 이벤트 + → 이후 새 요청 등록 금지, 입력 해시 변경 시 실행 금지 +``` + +응답 원장은 DB 밖에 두며 전송 시각과 응답 수신 시각은 해당 이벤트를 기록한 시계 기준이다. cutoff는 제어기가 입력을 고정한 시각이다. 실제 장애 발생·원본 접근 차단 시각과 같다고 추정하면 안 된다. 실제 시험은 두 시각과 차이를 별도 기록한다. + +## 복원 판정과 최초 감사 + +`dr_recovery.py`는 신규 실행 디렉터리만 만들고, 해시 연결 JSONL 원장에 순서·실행 ID·UTC 시각을 기록한다. 이는 누락·변경 검출을 위한 체크이며, 원장을 재작성할 수 있는 공격자에 대한 전자서명은 아니다. + +```text +그림 2 — _execute 내부의 변경 전 관문 [구현] + +execute.lock 획득 → 여섯 연산자·timeout·정리 명령 전체 검증 + │ + ▼ +restore → 고정 입력 해시 일치 + source_access_used=false 확인 + │ + ▼ +snapshot → Worker/Flyway/수선/원본 접근이 모두 false인지 검사 + │ → 요청마다 DB·버전·파일·Job ID 및 관계 대조 + │ → 보존 객체의 generation·실제 SHA-256·크기 대조 + ▼ +initial-snapshot.json + initial-audit.json 단독 생성·fsync + │ + ├─ 기능 코호트 누락 / 부분 DB / 원본 누락 / 해시 불일치 + │ → activate 호출 금지 → cleanup + └─ 기능 코호트 복원 + 손실 코호트의 DB 누락만 허용 + → activate → 격리 DB 버킷 재지정·Flyway·Worker 시작 허용 +``` + +`DB_MISSING`은 실제 조회가 누락을 확인한 경우에만 기록한다. 관측 행 자체가 없으면 증거 부족으로 중단한다. 최초 감사에 손실이 있더라도 후속 수선 결과로 이를 덮어쓰지 않는다. `generation`은 원본 버킷의 과거 숫자와 비교하는 것이 아니라 **동결한 보존 객체의 버전**과 비교한다. 별도 버킷 복사 시 새 generation을 동결 원장에 기록해야 한다. + +## 기능별 시간과 정리 + +```text +그림 3 — 최초 감사 이후와 종료 [구현·실행 설계] + +activate 완료 시각 + → functional 연산자 + ├─ download: 기존 원본 다운로드 + ├─ search: 기존 문서 검색 + ├─ job_resume: 중단 Job 재개 + ├─ new_upload: 새 업로드 + ├─ new_indexing: 새 문서 INDEXED + └─ new_download: 새 원본 다운로드 + 각 항목: 통과 여부·표본 수·관측 UTC·cutoff 이후 초 + → final_snapshot + → 미완료 Job·중복 Chunk·중복 Embedding·누락 Embedding 모두 0 + → final-audit 별도 생성 (initial-audit 보존) + → finally cleanup: 공개 설정 보존·원본 클러스터 복귀·소유 자원 정리 + ├─ 정리 확인 + 필수 단계 완료: PASS + └─ 단계 실패·중단·정리 미확인: INCOMPLETE +``` + +단계별 연산자 명령은 비공개 JSON의 argv 배열로만 전달한다. stdout JSON은 메모리에서 검증하며, stderr·명령문·SSH 주소·자격 증명을 증거에 복사하지 않는다. 스냅샷은 필드와 값 형식을 허용 목록으로 검사한 후 저장한다. 제어기는 연산자의 진술이 실제 환경과 같은지 독립적으로 증명하지 않는다. 후속 실행에서 DB SQL·보존 객체 읽기·HTTP 결과와 함께 근거를 남겨야 한다. + +## 사용 절차와 실행 전 조건 + +```bash +python3 scripts/opensql/dr_recovery.py init --run-dir "$RUN_DIR" --run-id "$RUN_ID" --revision "$REVISION" +python3 scripts/opensql/dr_recovery.py sent --run-dir "$RUN_DIR" --input "$SENT_JSON" +python3 scripts/opensql/dr_recovery.py response --run-dir "$RUN_DIR" --input "$RESPONSE_JSON" +python3 scripts/opensql/dr_recovery.py preflight --input "$INPUT_MANIFEST" +python3 scripts/opensql/dr_recovery.py freeze --run-dir "$RUN_DIR" --input "$INPUT_MANIFEST" +python3 scripts/opensql/dr_recovery.py execute --run-dir "$RUN_DIR" --input "$PRIVATE_OPERATOR_PLAN" +``` + +`RUN_DIR`는 존재하지 않는 새 디렉터리이며 부모는 준비되어 있어야 한다. 실제 실행 계획은 `restore`, `snapshot`, `activate`, `functional`, `final_snapshot`, `cleanup`을 모두 포함하고, 각 항목에 `argv`와 최대 1,800초의 `timeout_seconds`를 둔다. private plan 자체는 커밋하지 않는다. 같은 실행을 재시도하지 말고 새로운 run ID·PGDATA·원장으로 시작한다. + +준비 조건은 기준 백업 확인, WAL 연결 확인, 실제 수신 정상, 외부 identity 확보, 복원 대상 격리, 원복 준비, 공개 설정 보존이다. 모두 참인 boolean 입력은 실제 관측을 대신하지 않는다. 각 조건의 실제 근거가 없으면 false로 남기고 NO-GO를 기록한다. 이번 변경은 서비스 환경 파일·공개 버킷·운영 DB 정책을 바꾸지 않는다. diff --git a/docs/test-results/dr-upload-recovery-contract-20261018/README.md b/docs/test-results/dr-upload-recovery-contract-20261018/README.md new file mode 100644 index 00000000..e3075067 --- /dev/null +++ b/docs/test-results/dr-upload-recovery-contract-20261018/README.md @@ -0,0 +1,44 @@ +# 업로드 복원 판정 계약 검증 — 2026-10-18 + +## 목적 + +업로드 응답 원장을 복원 DB·원본과 대조하고, 최초 상태 감사 전에 Worker가 켜지지 않는지 검증한다. 이번 기록은 로컬 Python의 합성 JSON·제어 자식 프로세스 시험이다. 실제 OpenSQL·GCS·BGE-M3 복구를 실행한 결과가 아니다. + +## 테스트 환경 + +- 기반 리비전: `4cfe9666ac3a670bb72c8cd4d933bc7d7238fce0`와 이 PR의 계약 코드. +- 명령: `python3 -m unittest discover -s scripts/opensql -p test_dr_recovery.py -v` +- 외부 서비스: DB·Docker·GCP·모델 사용 안 함. 제어 연산자는 합성 JSON을 반환하는 로컬 Python 자식 프로세스. +- 원래 개인 환경 파일과 공개 서비스 설정 수정 없음. 파일 경로는 공개 로그 저장 전에 마스킹했다. + +## 테스트 방법 + +정상·불명 HTTP 결과, 중간 성공 응답 유실, DB 관계 불일치, 보존 객체 누락·해시·generation 불일치, 최초 감사 순서, 정리 누락을 각각 주입했다. 합격 기준은 판정 계약 전체 통과, 감사 미통과 시 activate 미호출, 정리 실패 시 PASS 금지다. + +## 관측 결과 + +| 실행 ID / 근거 | 시작 시각 (UTC, +00:00) | 표본·결과 | 해석 | +|---|---|---|---| +| [DR-CONTRACT-20261018-R1](runs/contract-r1/result.json) / [로그](runs/contract-r1/test.log) | 2026-10-18 08:13:06 | 26개: 성공 21·실패 3·오류 2, 종료 1 | 자식 연산자 시험 준비 오류. 미통과 회차로 보존 | +| [DR-CONTRACT-20261018-R2](runs/contract-r2/result.json) / [로그](runs/contract-r2/test.log) | 2026-10-18 08:13:55 | 26개: 성공 26·실패 0·오류 0, 종료 0 | 연산자 파일명 수정 후 재실행 통과 | +| [DR-CONTRACT-20261018-R3](runs/contract-r3/result.json) / [로그](runs/contract-r3/test.log) | 2026-10-18 08:21:28 | 33개: 성공 33·실패 0·오류 0, 0.689초, 종료 0 | 증거 필드 허용 목록·고정 객체 generation 대조 포함 로컬 통과 | +| [DR-CONTRACT-20261018-R4](runs/contract-r4/result.json) / [로그](runs/contract-r4/test.log) | 2026-10-18 08:27:14 | 36개: 성공 36·실패 0·오류 0, 0.769초, 종료 0 | 최종 감사의 원본 접근·잘못된 기능 응답 기록·감사 후 경합 차단 포함 최종 로컬 통과 | + +위 UTC 시작 시각의 한국 시각은 각각 17:13:06·17:13:55·17:21:28·17:27:14 KST(+09:00)다. 각 회차의 종료 시각은 result.json에 별도 기록했다. + +확인한 동작: + +- DB에서 사라진 중간 ACK 1건은 이후 ACK가 복원됐어도 손실 집계에서 빠지지 않는다. +- HTTP 500·응답 미수신과 실제 DB 복원 존재 여부를 분리한다. +- 최초 감사에 Worker/Flyway/수선/원본 접근 흔적이 있으면 기동 금지. +- 처음에 ACK 1건이 누락됐고 나중에 복구된 합성 회차에서도 초기 손실 1건을 보존한다. +- 고정 보존 객체 generation과 다른 객체는 해시가 같아도 최초 감사가 기동을 허용하지 않는다. +- 정리 명령 누락·잘못된 후속 명령은 복원 시작 전에 거부한다. 정리 실패와 실행 중단은 INCOMPLETE다. + +## 검증 과정에서 확인된 사항 + +첫 회차에서 자식 파일을 `operator.py`로 만든 것이 Python 표준 `operator` 모듈을 가렸다. `functools` import 순환 때문에 연산자 실행 5개가 실패·오류로 끝났다. 파일명을 `controlled_driver.py`로 바꾼 뒤 R2를 별도 실행했다. R3는 복원 입력 해시·민감 필드·보존 객체 버전 조건을 보강한 별도 회차다. R4는 최종 감사에서도 원본 접근을 금지하고, 거부된 기능 응답이 결과 파일에 남지 않으며 감사 뒤 늦은 응답이 합격 판정에 섞이지 않는지 검증했다. + +제어기가 받는 `source_access_used=false` 등은 연산자 확인값이다. 이 단위 시험은 실제 VM의 원본 접근 차단, 실제 WAL 연속성, OpenSQL 라이선스 복원 가능성, 검색 품질, 실제 복구 시간, RPO 또는 공개 서비스 전환을 증명하지 않는다. 후속 실험에서 각 조건과 연산자를 실제로 검증해야 한다. + +설계와 실제 실행은 [업로드 복원 판정 설계](../../design/issue-560-upload-recovery-contract.md)에서 구분한다. 초기 유실 집계와 최종 기능 수렴이 서로 다른 기록이라는 점을 유지한다. diff --git a/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r1/result.json b/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r1/result.json new file mode 100644 index 00000000..e627a6e1 --- /dev/null +++ b/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r1/result.json @@ -0,0 +1,15 @@ +{ + "run_id": "DR-CONTRACT-20261018-R1", + "purpose": "Synthetic contract and isolated child-process lifecycle", + "location": "local Python; no Docker/GCP/DB", + "started_at_utc": "2026-10-18T08:13:06.433309+00:00", + "ended_at_utc": "2026-10-18T08:13:06.717076+00:00", + "exit_code": 1, + "command": "python3 -m unittest discover -s scripts/opensql -p test_dr_recovery.py -v", + "revision": "4cfe9666ac3a670bb72c8cd4d933bc7d7238fce0 + submitted recovery contract changes", + "test_count": 26, + "passed": 21, + "failures": 3, + "errors": 2, + "verdict": "FAIL" +} diff --git a/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r1/test.log b/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r1/test.log new file mode 100644 index 00000000..59d33f4d --- /dev/null +++ b/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r1/test.log @@ -0,0 +1,94 @@ +test_audit_failure_never_activates_and_cleans_up (test_dr_recovery.RecoveryContractTest) ... ERROR +test_bad_later_command_blocks_restore (test_dr_recovery.RecoveryContractTest) ... ok +test_changed_inputs_block_operator (test_dr_recovery.RecoveryContractTest) ... ok +test_cleanup_failure_is_not_pass (test_dr_recovery.RecoveryContractTest) ... FAIL +test_corrupt_ledger_is_not_summarized (test_dr_recovery.RecoveryContractTest) ... ok +test_duplicate_response_rejected (test_dr_recovery.RecoveryContractTest) ... ok +test_duplicate_wal_rejected (test_dr_recovery.RecoveryContractTest) ... ok +test_finished_run_cannot_be_executed_again (test_dr_recovery.RecoveryContractTest) ... ok +test_frozen_inputs_cannot_be_replaced (test_dr_recovery.RecoveryContractTest) ... ok +test_http_error_may_be_present_in_restored_db (test_dr_recovery.RecoveryContractTest) ... ok +test_initial_loss_preserved_when_final_snapshot_is_repaired (test_dr_recovery.RecoveryContractTest) ... FAIL +test_initial_state_rejects_started_worker_or_flyway (test_dr_recovery.RecoveryContractTest) ... ok +test_inputs_missing_check_is_no_go (test_dr_recovery.RecoveryContractTest) ... ok +test_missing_cleanup_blocks_all_actions (test_dr_recovery.RecoveryContractTest) ... ok +test_missing_observation_is_not_db_loss (test_dr_recovery.RecoveryContractTest) ... ok +test_missing_response_is_unknown_not_failure (test_dr_recovery.RecoveryContractTest) ... ok +test_object_missing_blocks_worker (test_dr_recovery.RecoveryContractTest) ... ok +test_old_and_new_success_do_not_hide_middle_loss (test_dr_recovery.RecoveryContractTest) ... ok +test_original_identity_must_be_preserved (test_dr_recovery.RecoveryContractTest) ... ok +test_partial_wal_rejected (test_dr_recovery.RecoveryContractTest) ... ok +test_real_hash_mismatch_is_not_restored (test_dr_recovery.RecoveryContractTest) ... ok +test_response_ledger_is_durable_and_private (test_dr_recovery.RecoveryContractTest) ... ok +test_restore_failure_cleans_up (test_dr_recovery.RecoveryContractTest) ... ERROR +test_success_requires_all_four_ids (test_dr_recovery.RecoveryContractTest) ... ok +test_success_response_ids_must_match_restored_relations (test_dr_recovery.RecoveryContractTest) ... ok +test_success_stages_audit_before_activation_then_final_audit (test_dr_recovery.RecoveryContractTest) ... FAIL + +====================================================================== +ERROR: test_audit_failure_never_activates_and_cleans_up (test_dr_recovery.RecoveryContractTest) +---------------------------------------------------------------------- +Traceback (most recent call last): + File "", line 204, in test_audit_failure_never_activates_and_cleans_up + self.assertEqual((self.root/'calls.txt').read_text().splitlines(), ['restore', 'snapshot', 'cleanup']) + File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/pathlib.py", line 1256, in read_text + with self.open(mode='r', encoding=encoding, errors=errors) as f: + File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/pathlib.py", line 1242, in open + return io.open(self, mode, buffering, encoding, errors, newline, + File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/pathlib.py", line 1110, in _opener + return self._accessor.open(self, flags, mode) +FileNotFoundError: [Errno 2] No such file or directory: ' + +====================================================================== +ERROR: test_restore_failure_cleans_up (test_dr_recovery.RecoveryContractTest) +---------------------------------------------------------------------- +Traceback (most recent call last): + File "", line 210, in test_restore_failure_cleans_up + self.assertEqual((self.root/'calls.txt').read_text().splitlines(), ['restore', 'cleanup']) + File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/pathlib.py", line 1256, in read_text + with self.open(mode='r', encoding=encoding, errors=errors) as f: + File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/pathlib.py", line 1242, in open + return io.open(self, mode, buffering, encoding, errors, newline, + File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/pathlib.py", line 1110, in _opener + return self._accessor.open(self, flags, mode) +FileNotFoundError: [Errno 2] No such file or directory: ' + +====================================================================== +FAIL: test_cleanup_failure_is_not_pass (test_dr_recovery.RecoveryContractTest) +---------------------------------------------------------------------- +Traceback (most recent call last): + File "", line 227, in test_cleanup_failure_is_not_pass + self.assertEqual(result['reason'], 'OPERATOR_CLEANUP_FAILED') +AssertionError: 'OPERATOR_RESTORE_FAILED' != 'OPERATOR_CLEANUP_FAILED' +- OPERATOR_RESTORE_FAILED +? ^ ^^^^^ ++ OPERATOR_CLEANUP_FAILED +? ^^ ^^^^ + + +====================================================================== +FAIL: test_initial_loss_preserved_when_final_snapshot_is_repaired (test_dr_recovery.RecoveryContractTest) +---------------------------------------------------------------------- +Traceback (most recent call last): + File "", line 196, in test_initial_loss_preserved_when_final_snapshot_is_repaired + self.assertEqual(result['execution'], 'PASS') +AssertionError: 'INCOMPLETE' != 'PASS' +- INCOMPLETE ++ PASS + + +====================================================================== +FAIL: test_success_stages_audit_before_activation_then_final_audit (test_dr_recovery.RecoveryContractTest) +---------------------------------------------------------------------- +Traceback (most recent call last): + File "", line 185, in test_success_stages_audit_before_activation_then_final_audit + self.assertEqual(result["execution"], "PASS") +AssertionError: 'INCOMPLETE' != 'PASS' +- INCOMPLETE ++ PASS + + +---------------------------------------------------------------------- +Ran 26 tests in 0.236s + +FAILED (failures=3, errors=2) diff --git a/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r2/result.json b/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r2/result.json new file mode 100644 index 00000000..42b4af8b --- /dev/null +++ b/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r2/result.json @@ -0,0 +1,15 @@ +{ + "run_id": "DR-CONTRACT-20261018-R2", + "purpose": "Contract rerun after child-driver module naming fix", + "location": "local Python; no Docker/GCP/DB", + "started_at_utc": "2026-10-18T08:13:55.061338+00:00", + "ended_at_utc": "2026-10-18T08:13:55.649096+00:00", + "exit_code": 0, + "command": "python3 -m unittest discover -s scripts/opensql -p test_dr_recovery.py -v", + "revision": "4cfe9666ac3a670bb72c8cd4d933bc7d7238fce0 + submitted recovery contract changes", + "test_count": 26, + "passed": 26, + "failures": 0, + "errors": 0, + "verdict": "PASS" +} diff --git a/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r2/test.log b/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r2/test.log new file mode 100644 index 00000000..bb3053db --- /dev/null +++ b/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r2/test.log @@ -0,0 +1,31 @@ +test_audit_failure_never_activates_and_cleans_up (test_dr_recovery.RecoveryContractTest) ... ok +test_bad_later_command_blocks_restore (test_dr_recovery.RecoveryContractTest) ... ok +test_changed_inputs_block_operator (test_dr_recovery.RecoveryContractTest) ... ok +test_cleanup_failure_is_not_pass (test_dr_recovery.RecoveryContractTest) ... ok +test_corrupt_ledger_is_not_summarized (test_dr_recovery.RecoveryContractTest) ... ok +test_duplicate_response_rejected (test_dr_recovery.RecoveryContractTest) ... ok +test_duplicate_wal_rejected (test_dr_recovery.RecoveryContractTest) ... ok +test_finished_run_cannot_be_executed_again (test_dr_recovery.RecoveryContractTest) ... ok +test_frozen_inputs_cannot_be_replaced (test_dr_recovery.RecoveryContractTest) ... ok +test_http_error_may_be_present_in_restored_db (test_dr_recovery.RecoveryContractTest) ... ok +test_initial_loss_preserved_when_final_snapshot_is_repaired (test_dr_recovery.RecoveryContractTest) ... ok +test_initial_state_rejects_started_worker_or_flyway (test_dr_recovery.RecoveryContractTest) ... ok +test_inputs_missing_check_is_no_go (test_dr_recovery.RecoveryContractTest) ... ok +test_missing_cleanup_blocks_all_actions (test_dr_recovery.RecoveryContractTest) ... ok +test_missing_observation_is_not_db_loss (test_dr_recovery.RecoveryContractTest) ... ok +test_missing_response_is_unknown_not_failure (test_dr_recovery.RecoveryContractTest) ... ok +test_object_missing_blocks_worker (test_dr_recovery.RecoveryContractTest) ... ok +test_old_and_new_success_do_not_hide_middle_loss (test_dr_recovery.RecoveryContractTest) ... ok +test_original_identity_must_be_preserved (test_dr_recovery.RecoveryContractTest) ... ok +test_partial_wal_rejected (test_dr_recovery.RecoveryContractTest) ... ok +test_real_hash_mismatch_is_not_restored (test_dr_recovery.RecoveryContractTest) ... ok +test_response_ledger_is_durable_and_private (test_dr_recovery.RecoveryContractTest) ... ok +test_restore_failure_cleans_up (test_dr_recovery.RecoveryContractTest) ... ok +test_success_requires_all_four_ids (test_dr_recovery.RecoveryContractTest) ... ok +test_success_response_ids_must_match_restored_relations (test_dr_recovery.RecoveryContractTest) ... ok +test_success_stages_audit_before_activation_then_final_audit (test_dr_recovery.RecoveryContractTest) ... ok + +---------------------------------------------------------------------- +Ran 26 tests in 0.544s + +OK diff --git a/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r3/result.json b/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r3/result.json new file mode 100644 index 00000000..3ccf537f --- /dev/null +++ b/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r3/result.json @@ -0,0 +1,16 @@ +{ + "run_id": "DR-CONTRACT-20261018-R3", + "purpose": "고정 객체 버전과 최초 감사·Worker 기동 순서 계약", + "environment": "local Python synthetic input and controlled child commands", + "revision": "4cfe9666ac3a670bb72c8cd4d933bc7d7238fce0 + submitted recovery contract changes", + "command": "python3 -m unittest discover -s scripts/opensql -p test_dr_recovery.py -v", + "started_at_utc": "2026-10-18T08:21:28.051523+00:00", + "finished_at_utc": "2026-10-18T08:21:28.740579+00:00", + "duration_seconds": 0.689, + "test_count": 33, + "exit_code": 0, + "verdict": "PASS", + "passed": 33, + "failures": 0, + "errors": 0 +} diff --git a/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r3/test.log b/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r3/test.log new file mode 100644 index 00000000..e5f9576b --- /dev/null +++ b/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r3/test.log @@ -0,0 +1,38 @@ +test_audit_failure_never_activates_and_cleans_up (test_dr_recovery.RecoveryContractTest) ... ok +test_bad_later_command_blocks_restore (test_dr_recovery.RecoveryContractTest) ... ok +test_changed_inputs_block_operator (test_dr_recovery.RecoveryContractTest) ... ok +test_cleanup_failure_is_not_pass (test_dr_recovery.RecoveryContractTest) ... ok +test_corrupt_ledger_is_not_summarized (test_dr_recovery.RecoveryContractTest) ... ok +test_duplicate_response_rejected (test_dr_recovery.RecoveryContractTest) ... ok +test_duplicate_wal_rejected (test_dr_recovery.RecoveryContractTest) ... ok +test_finished_run_cannot_be_executed_again (test_dr_recovery.RecoveryContractTest) ... ok +test_frozen_inputs_cannot_be_replaced (test_dr_recovery.RecoveryContractTest) ... ok +test_http_error_may_be_present_in_restored_db (test_dr_recovery.RecoveryContractTest) ... ok +test_initial_loss_preserved_when_final_snapshot_is_repaired (test_dr_recovery.RecoveryContractTest) ... ok +test_initial_state_rejects_started_worker_or_flyway (test_dr_recovery.RecoveryContractTest) ... ok +test_inputs_missing_check_is_no_go (test_dr_recovery.RecoveryContractTest) ... ok +test_missing_cleanup_blocks_all_actions (test_dr_recovery.RecoveryContractTest) ... ok +test_missing_observation_is_not_db_loss (test_dr_recovery.RecoveryContractTest) ... ok +test_missing_pinned_originals_is_no_go (test_dr_recovery.RecoveryContractTest) ... ok +test_missing_response_is_unknown_not_failure (test_dr_recovery.RecoveryContractTest) ... ok +test_object_missing_blocks_worker (test_dr_recovery.RecoveryContractTest) ... ok +test_old_and_new_success_do_not_hide_middle_loss (test_dr_recovery.RecoveryContractTest) ... ok +test_original_identity_must_be_preserved (test_dr_recovery.RecoveryContractTest) ... ok +test_partial_wal_rejected (test_dr_recovery.RecoveryContractTest) ... ok +test_private_consistency_field_is_rejected (test_dr_recovery.RecoveryContractTest) ... ok +test_private_extra_row_fields_are_rejected_before_recording (test_dr_recovery.RecoveryContractTest) ... ok +test_private_extra_snapshot_fields_are_rejected_before_recording (test_dr_recovery.RecoveryContractTest) ... ok +test_private_value_in_missing_row_is_rejected (test_dr_recovery.RecoveryContractTest) ... ok +test_real_hash_mismatch_is_not_restored (test_dr_recovery.RecoveryContractTest) ... ok +test_response_ledger_is_durable_and_private (test_dr_recovery.RecoveryContractTest) ... ok +test_restore_failure_cleans_up (test_dr_recovery.RecoveryContractTest) ... ok +test_restore_receipt_must_bind_frozen_manifest (test_dr_recovery.RecoveryContractTest) ... ok +test_success_requires_all_four_ids (test_dr_recovery.RecoveryContractTest) ... ok +test_success_response_ids_must_match_restored_relations (test_dr_recovery.RecoveryContractTest) ... ok +test_success_stages_audit_before_activation_then_final_audit (test_dr_recovery.RecoveryContractTest) ... ok +test_wrong_preserved_generation_blocks_activation (test_dr_recovery.RecoveryContractTest) ... ok + +---------------------------------------------------------------------- +Ran 33 tests in 0.641s + +OK diff --git a/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r4/result.json b/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r4/result.json new file mode 100644 index 00000000..c0f54894 --- /dev/null +++ b/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r4/result.json @@ -0,0 +1,16 @@ +{ + "run_id": "DR-CONTRACT-20261018-R4", + "purpose": "최종 감사 원본 의존·잘못된 응답 기록·감사 이후 경합 차단", + "environment": "local Python synthetic input and controlled child commands", + "revision": "4cfe9666ac3a670bb72c8cd4d933bc7d7238fce0 + submitted recovery contract changes", + "command": "python3 -m unittest discover -s scripts/opensql -p test_dr_recovery.py -v", + "started_at_utc": "2026-10-18T08:27:14.595070+00:00", + "finished_at_utc": "2026-10-18T08:27:15.364571+00:00", + "duration_seconds": 0.769, + "test_count": 36, + "exit_code": 0, + "verdict": "PASS", + "passed": 36, + "failures": 0, + "errors": 0 +} diff --git a/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r4/test.log b/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r4/test.log new file mode 100644 index 00000000..1a8551fe --- /dev/null +++ b/docs/test-results/dr-upload-recovery-contract-20261018/runs/contract-r4/test.log @@ -0,0 +1,41 @@ +test_audit_failure_never_activates_and_cleans_up (test_dr_recovery.RecoveryContractTest) ... ok +test_bad_later_command_blocks_restore (test_dr_recovery.RecoveryContractTest) ... ok +test_changed_inputs_block_operator (test_dr_recovery.RecoveryContractTest) ... ok +test_cleanup_failure_is_not_pass (test_dr_recovery.RecoveryContractTest) ... ok +test_corrupt_ledger_is_not_summarized (test_dr_recovery.RecoveryContractTest) ... ok +test_duplicate_response_rejected (test_dr_recovery.RecoveryContractTest) ... ok +test_duplicate_wal_rejected (test_dr_recovery.RecoveryContractTest) ... ok +test_final_snapshot_cannot_use_original_cluster (test_dr_recovery.RecoveryContractTest) ... ok +test_finished_run_cannot_be_executed_again (test_dr_recovery.RecoveryContractTest) ... ok +test_frozen_inputs_cannot_be_replaced (test_dr_recovery.RecoveryContractTest) ... ok +test_http_error_may_be_present_in_restored_db (test_dr_recovery.RecoveryContractTest) ... ok +test_initial_loss_preserved_when_final_snapshot_is_repaired (test_dr_recovery.RecoveryContractTest) ... ok +test_initial_state_rejects_started_worker_or_flyway (test_dr_recovery.RecoveryContractTest) ... ok +test_inputs_missing_check_is_no_go (test_dr_recovery.RecoveryContractTest) ... ok +test_invalid_functional_receipt_is_not_persisted (test_dr_recovery.RecoveryContractTest) ... ok +test_missing_cleanup_blocks_all_actions (test_dr_recovery.RecoveryContractTest) ... ok +test_missing_observation_is_not_db_loss (test_dr_recovery.RecoveryContractTest) ... ok +test_missing_pinned_originals_is_no_go (test_dr_recovery.RecoveryContractTest) ... ok +test_missing_response_is_unknown_not_failure (test_dr_recovery.RecoveryContractTest) ... ok +test_object_missing_blocks_worker (test_dr_recovery.RecoveryContractTest) ... ok +test_old_and_new_success_do_not_hide_middle_loss (test_dr_recovery.RecoveryContractTest) ... ok +test_original_identity_must_be_preserved (test_dr_recovery.RecoveryContractTest) ... ok +test_partial_wal_rejected (test_dr_recovery.RecoveryContractTest) ... ok +test_private_consistency_field_is_rejected (test_dr_recovery.RecoveryContractTest) ... ok +test_private_extra_row_fields_are_rejected_before_recording (test_dr_recovery.RecoveryContractTest) ... ok +test_private_extra_snapshot_fields_are_rejected_before_recording (test_dr_recovery.RecoveryContractTest) ... ok +test_private_value_in_missing_row_is_rejected (test_dr_recovery.RecoveryContractTest) ... ok +test_racing_response_after_audit_is_rejected_when_replayed (test_dr_recovery.RecoveryContractTest) ... ok +test_real_hash_mismatch_is_not_restored (test_dr_recovery.RecoveryContractTest) ... ok +test_response_ledger_is_durable_and_private (test_dr_recovery.RecoveryContractTest) ... ok +test_restore_failure_cleans_up (test_dr_recovery.RecoveryContractTest) ... ok +test_restore_receipt_must_bind_frozen_manifest (test_dr_recovery.RecoveryContractTest) ... ok +test_success_requires_all_four_ids (test_dr_recovery.RecoveryContractTest) ... ok +test_success_response_ids_must_match_restored_relations (test_dr_recovery.RecoveryContractTest) ... ok +test_success_stages_audit_before_activation_then_final_audit (test_dr_recovery.RecoveryContractTest) ... ok +test_wrong_preserved_generation_blocks_activation (test_dr_recovery.RecoveryContractTest) ... ok + +---------------------------------------------------------------------- +Ran 36 tests in 0.724s + +OK diff --git a/scripts/opensql/dr_recovery.py b/scripts/opensql/dr_recovery.py new file mode 100755 index 00000000..0ed264c6 --- /dev/null +++ b/scripts/opensql/dr_recovery.py @@ -0,0 +1,499 @@ +#!/usr/bin/env python3 +"""Audit upload acknowledgements before activating an isolated restored service. + +Barman and GCS retain ownership of backup creation and transfer. This experiment +controller owns only its fresh evidence directory and explicitly supplied private +operator commands; it never discovers credentials or modifies public settings. +""" +from __future__ import annotations + +import argparse +from datetime import datetime, timezone +import hashlib +import json +import os +from pathlib import Path +import re +import subprocess +import sys +import time + +from ha_evidence import json_text, replace_private, write_private + +LABEL = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,99}\Z") +SHA = re.compile(r"[0-9a-f]{64}\Z") +WAL = re.compile(r"(?:[0-9A-F]{24}|[0-9A-F]{8}\.history)\Z") +IDS = ("documentId", "documentVersionId", "fileObjectId", "embeddingJobId") +CHECKS = ("backup_verified", "wal_chain_verified", "receiver_healthy", + "identity_available_offsite", "restore_target_isolated", "rollback_ready", + "public_config_preserved") +FUNCTIONS = ("download", "search", "job_resume", "new_upload", "new_indexing", "new_download") + + +class ContractError(ValueError): + """Reject incomplete evidence without exposing private operator output.""" + + +def utc_now(): + return datetime.now(timezone.utc).isoformat(timespec="milliseconds") + + +def instant(value): + try: + result = datetime.fromisoformat(value.replace("Z", "+00:00")) + if result.tzinfo is None: + raise ValueError() + return result + except (AttributeError, TypeError, ValueError): + raise ContractError("INVALID_TIMESTAMP") from None + + +def load(path): + try: + return json.loads(Path(path).read_text(encoding="utf-8")) + except (OSError, ValueError): + raise ContractError("INVALID_JSON") from None + + +def digest(value): + return hashlib.sha256(json_text(value).encode()).hexdigest() + + +def positive(value): + return type(value) is int and value > 0 + + +def initialize(directory, run_id, revision): + if not LABEL.fullmatch(run_id) or not re.fullmatch(r"[0-9a-f]{7,40}", revision): + raise ContractError("INVALID_RUN_IDENTITY") + directory.mkdir(mode=0o700, parents=False, exist_ok=False) + write_private(directory / "manifest.json", json_text({"schema_version": 1, + "run_id": run_id, "revision": revision, "created_at_utc": utc_now()}), exclusive=True) + write_private(directory / "events.jsonl", "", exclusive=True) + + +def events(directory): + manifest = load(directory / "manifest.json") + if manifest.get("schema_version") != 1 or not LABEL.fullmatch(manifest.get("run_id", "")): + raise ContractError("INVALID_MANIFEST") + result, previous = [], "0" * 64 + try: + lines = (directory / "events.jsonl").read_text().splitlines() + for ordinal, line in enumerate(lines, 1): + event = json.loads(line) + checksum = event.pop("sha256") + if (event.get("sequence") != ordinal or event.get("previous") != previous + or event.get("run_id") != manifest["run_id"] or digest(event) != checksum): + raise ValueError() + instant(event["at_utc"]) + event["sha256"] = checksum + result.append(event) + previous = checksum + except (OSError, ValueError, KeyError, TypeError): + raise ContractError("CORRUPT_EVENT_LEDGER") from None + return result + + +def append(directory, kind, data): + # One experiment controller serializes stages; flock also protects live response writers. + import fcntl + with (directory / "events.lock").open("a+") as lock: + os.fchmod(lock.fileno(), 0o600) + fcntl.flock(lock, fcntl.LOCK_EX) + old = events(directory) + if old and old[-1]["kind"] == "finished": + raise ContractError("RUN_ALREADY_FINISHED") + event = {"sequence": len(old) + 1, "run_id": load(directory / "manifest.json")["run_id"], + "previous": old[-1]["sha256"] if old else "0" * 64, + "kind": kind, "at_utc": utc_now(), "data": data} + event["sha256"] = digest(event) + with (directory / "events.jsonl").open("a", encoding="utf-8") as output: + output.write(json.dumps(event, ensure_ascii=False, sort_keys=True) + "\n") + output.flush() + os.fsync(output.fileno()) + return event + + +def record_sent(directory, record): + required = {"request_id", "sha256", "size", "cohort"} + if (set(record) != required or not LABEL.fullmatch(record.get("request_id", "")) + or not SHA.fullmatch(record.get("sha256", "")) or not positive(record.get("size")) + or record.get("cohort") not in ("functional", "loss")): + raise ContractError("INVALID_REQUEST") + old = events(directory) + if any(e["kind"] == "cutoff" for e in old): + raise ContractError("REQUEST_AFTER_CUTOFF") + if any(e["kind"] == "sent" and e["data"]["request_id"] == record["request_id"] for e in old): + raise ContractError("DUPLICATE_REQUEST") + append(directory, "sent", record) + + +def record_response(directory, record): + if set(record) - {"request_id", "http_status", "ids"} or "request_id" not in record: + raise ContractError("INVALID_RESPONSE") + old = events(directory) + request_id = record["request_id"] + if not any(e["kind"] == "sent" and e["data"]["request_id"] == request_id for e in old): + raise ContractError("RESPONSE_WITHOUT_REQUEST") + if any(e["kind"] == "response" and e["data"]["request_id"] == request_id for e in old): + raise ContractError("DUPLICATE_RESPONSE") + if any(e["kind"] == "initial_audit" for e in old): + raise ContractError("RESPONSE_AFTER_AUDIT") + status = record.get("http_status") + if status is not None and (type(status) is not int or not 100 <= status <= 599): + raise ContractError("INVALID_HTTP_STATUS") + if status in (200, 201): + if set(record.get("ids", {})) != set(IDS) or not all(positive(record["ids"][k]) for k in IDS): + raise ContractError("SUCCESS_WITHOUT_IDS") + elif record.get("ids"): + raise ContractError("IDS_WITHOUT_SUCCESS") + # HTTP errors remain observations. They never prove that a DB transaction rolled back. + append(directory, "response", {**record, "outcome": "ACKNOWLEDGED" if status in (200, 201) + else "UNKNOWN" if status is None else "HTTP_ERROR"}) + + +def validate_inputs(value): + if not isinstance(value, dict) or value.get("schema_version") != 1: + raise ContractError("INVALID_INPUT_SCHEMA") + if set(value) - {"schema_version", "captured_at_utc", "checks", "backup", "identity_sha256", "wals", "originals"}: + raise ContractError("UNSAFE_INPUT_FIELDS") + instant(value.get("captured_at_utc")) + checks = value.get("checks", {}) + if not isinstance(checks, dict) or set(checks) - set(CHECKS): + raise ContractError("UNSAFE_INPUT_FIELDS") + missing = [name for name in CHECKS if checks.get(name) is not True] + if missing: + return {"ready": False, "reasons": missing} + backup = value.get("backup", {}) + if not isinstance(backup, dict) or set(backup) != {"id", "sha256", "size", "generation"}: + raise ContractError("UNSAFE_INPUT_FIELDS") + if (not re.fullmatch(r"[0-9]{8}T[0-9]{6}", backup.get("id", "")) + or not SHA.fullmatch(backup.get("sha256", "")) or not positive(backup.get("size")) + or not str(backup.get("generation", "")).isdigit()): + raise ContractError("INVALID_BACKUP_IDENTITY") + names = set() + if not value.get("wals"): + raise ContractError("NO_ARCHIVED_WAL") + for wal in value["wals"]: + if not isinstance(wal, dict) or set(wal) != {"name", "sha256", "size", "generation"}: + raise ContractError("UNSAFE_INPUT_FIELDS") + name = wal.get("name", "") + if (not WAL.fullmatch(name) or name in names or not SHA.fullmatch(wal.get("sha256", "")) + or not positive(wal.get("size")) or not str(wal.get("generation", "")).isdigit()): + raise ContractError("INVALID_WAL_OBJECT") + names.add(name) + if not SHA.fullmatch(value.get("identity_sha256", "")): + raise ContractError("NO_OFFSITE_IDENTITY") + original_ids = set() + if not isinstance(value.get("originals"), list) or not value["originals"]: + raise ContractError("NO_PINNED_ORIGINALS") + for original in value["originals"]: + if (not isinstance(original, dict) or set(original) != {"fileObjectId", "sha256", "size", "generation"} + or not positive(original["fileObjectId"]) or original["fileObjectId"] in original_ids + or not SHA.fullmatch(original["sha256"]) or not positive(original["size"]) + or not str(original["generation"]).isdigit()): + raise ContractError("INVALID_ORIGINAL_OBJECT") + original_ids.add(original["fileObjectId"]) + return {"ready": True, "reasons": [], "wal_objects": len(names)} + + +def freeze_inputs(directory, value): + if any(e["kind"] == "cutoff" for e in events(directory)): + raise ContractError("INPUTS_ALREADY_FROZEN") + result = validate_inputs(value) + if not result["ready"]: + raise ContractError("PREFLIGHT_NO_GO") + if instant(value["captured_at_utc"]) > datetime.now(timezone.utc): + raise ContractError("FUTURE_INPUT_CAPTURE") + write_private(directory / "inputs.json", json_text(value), exclusive=True) + append(directory, "cutoff", {"inputs_sha256": digest(value), "wal_objects": result["wal_objects"]}) + + +def requests(directory): + sent, received, cutoff_seen, audit_seen = {}, {}, False, False + for event in events(directory): + cutoff_seen = cutoff_seen or event["kind"] == "cutoff" + audit_seen = audit_seen or event["kind"] == "initial_audit" + if event["kind"] == "sent": + if cutoff_seen: + raise ContractError("REQUEST_AFTER_CUTOFF") + key = event["data"]["request_id"] + if key in sent: + raise ContractError("DUPLICATE_REQUEST_EVIDENCE") + sent[key] = {**event["data"], "sent_at_utc": event["at_utc"]} + if event["kind"] == "response": + if audit_seen: + raise ContractError("RESPONSE_AFTER_AUDIT") + key = event["data"]["request_id"] + if key not in sent or key in received: + raise ContractError("INVALID_RESPONSE_EVIDENCE") + received[key] = {**event["data"], "ack_at_utc": event["at_utc"]} + return [{**record, **received.get(key, {"outcome": "UNKNOWN", "http_status": None})} + for key, record in sent.items()] + + +def audit_records(records, snapshot, *, pristine=True, originals=None): + """Compare original acknowledgements with immutable first-restoration observations.""" + if not isinstance(snapshot, dict): + raise ContractError("INVALID_SNAPSHOT") + if snapshot.get("source_access_used") is not False: + raise ContractError("SNAPSHOT_SOURCE_DEPENDENCY") + if pristine and any(snapshot.get(key) is not False for key in ("worker_started", "flyway_ran", "repair_ran", "source_access_used")): + raise ContractError("INITIAL_SNAPSHOT_NOT_PRISTINE") + allowed_snapshot = {"worker_started", "flyway_ran", "repair_ran", "source_access_used", "records", "consistency"} + if set(snapshot) - allowed_snapshot: + raise ContractError("UNSAFE_SNAPSHOT_FIELDS") + for name in ("worker_started", "flyway_ran", "repair_ran", "source_access_used"): + if type(snapshot.get(name)) is not bool: + raise ContractError("INVALID_SNAPSHOT_FLAGS") + counters = snapshot.get("consistency", {}) + allowed_counters = {"checked_documents", "incomplete_jobs", "duplicate_chunks", "duplicate_embeddings", "missing_embeddings"} + if (not isinstance(counters, dict) or set(counters) - allowed_counters + or any(type(v) is not int or v < 0 for v in counters.values())): + raise ContractError("UNSAFE_SNAPSHOT_FIELDS") + pinned = {row["fileObjectId"]: row for row in originals or []} + rows = snapshot.get("records") + if not isinstance(rows, list): + raise ContractError("INVALID_SNAPSHOT") + indexed = {} + allowed_row = {"request_id", "state", "ids", "relations_match", "sha256", "size", "generation", + "document_status", "version_status", "job_status", "attempt_count", "chunk_count", "embedding_count"} + for row in rows: + if not isinstance(row, dict) or set(row) - allowed_row: + raise ContractError("UNSAFE_SNAPSHOT_FIELDS") + if ("ids" in row and (not isinstance(row["ids"], dict) or set(row["ids"]) != set(IDS) + or not all(positive(v) for v in row["ids"].values()))): + raise ContractError("INVALID_SNAPSHOT_IDS") + if "relations_match" in row and type(row["relations_match"]) is not bool: + raise ContractError("INVALID_SNAPSHOT_RELATIONS") + if "size" in row and not positive(row["size"]): + raise ContractError("INVALID_SNAPSHOT_SIZE") + if "sha256" in row and (not isinstance(row["sha256"], str) or not SHA.fullmatch(row["sha256"])): + raise ContractError("INVALID_SNAPSHOT_HASH") + if "generation" in row and (not isinstance(row["generation"], (int, str)) or not str(row["generation"]).isdigit()): + raise ContractError("MISSING_OBJECT_GENERATION") + for key in ("attempt_count", "chunk_count", "embedding_count"): + if key in row and (type(row[key]) is not int or row[key] < 0): + raise ContractError("INVALID_SNAPSHOT_COUNTER") + for key in ("document_status", "version_status", "job_status"): + if key in row and row[key] not in ("UPLOADED", "PENDING", "PROCESSING", "INDEXED", "FAILED", "DELETED"): + raise ContractError("INVALID_SNAPSHOT_STATUS") + key = row.get("request_id") + if key in indexed or key not in {r["request_id"] for r in records}: + raise ContractError("UNBOUND_OR_DUPLICATE_SNAPSHOT") + indexed[key] = row + output = [] + for request in records: + row = indexed.get(request["request_id"]) + outcome = request["outcome"] + if row is None: + raise ContractError("MISSING_SNAPSHOT_OBSERVATION") + state = row.get("state") + if state not in ("DB_MISSING", "DB_PARTIAL", "OBJECT_MISSING", "HASH_MISMATCH", "RESTORED"): + raise ContractError("INVALID_RESTORED_STATE") + if state == "RESTORED": + observed_ids = row.get("ids", {}) + if set(observed_ids) != set(IDS) or not all(positive(observed_ids[k]) for k in IDS): + raise ContractError("INCOMPLETE_RESTORED_IDS") + if outcome == "ACKNOWLEDGED" and observed_ids != request["ids"]: + raise ContractError("RESPONSE_BINDING_MISMATCH") + if row.get("relations_match") is not True: + raise ContractError("BROKEN_DOCUMENT_RELATIONS") + if row.get("sha256") != request["sha256"] or row.get("size") != request["size"]: + state = "HASH_MISMATCH" + if not str(row.get("generation", "")).isdigit(): + raise ContractError("MISSING_OBJECT_GENERATION") + # Pin the preserved object's generation; an unrelated live object cannot satisfy restoration. + if originals is not None: + original = pinned.get(observed_ids["fileObjectId"]) + if not original or any(str(row[k]) != str(original[k]) for k in ("sha256", "size", "generation")): + raise ContractError("RESTORED_OBJECT_NOT_PINNED") + output.append({"request_id": request["request_id"], "cohort": request["cohort"], + "http_outcome": outcome, "restored_state": state, + "ack_at_utc": request.get("ack_at_utc")}) + acknowledged = [r for r in output if r["http_outcome"] == "ACKNOWLEDGED"] + missing = [r for r in acknowledged if r["restored_state"] != "RESTORED"] + activation_allowed = (bool(records) and any(r["cohort"] == "functional" for r in output) + and all(r["restored_state"] == "RESTORED" for r in output if r["cohort"] == "functional") + and not any(r["restored_state"] in ("DB_PARTIAL", "OBJECT_MISSING", "HASH_MISMATCH") for r in output)) + return {"acknowledged": len(acknowledged), "restored_acknowledged": len(acknowledged) - len(missing), + "lost_or_incomplete_acknowledged": len(missing), + "unknown_http": sum(r["http_outcome"] == "UNKNOWN" for r in output), + "http_error": sum(r["http_outcome"] == "HTTP_ERROR" for r in output), + "activation_allowed": activation_allowed, "requests": output} + + +def initial_audit(directory, snapshot): + old = events(directory) + if not any(e["kind"] == "restored" for e in old) or any(e["kind"] == "initial_audit" for e in old): + raise ContractError("INVALID_AUDIT_ORDER") + result = audit_records(requests(directory), snapshot, originals=load(directory / "inputs.json")["originals"]) + write_private(directory / "initial-snapshot.json", json_text(snapshot), exclusive=True) + write_private(directory / "initial-audit.json", json_text(result), exclusive=True) + append(directory, "initial_audit", {"snapshot_sha256": digest(snapshot), "audit_sha256": digest(result), + "activation_allowed": result["activation_allowed"], "acknowledged": result["acknowledged"], + "lost_or_incomplete_acknowledged": result["lost_or_incomplete_acknowledged"]}) + return result + + +def execute_operator(plan, name): + action = plan.get(name, {}) + command = action.get("argv") + timeout = action.get("timeout_seconds", 300) + if not isinstance(command, list) or not command or not all(isinstance(x, str) for x in command): + raise ContractError("INVALID_OPERATOR_COMMAND") + if type(timeout) is not int or not 1 <= timeout <= 1800: + raise ContractError("INVALID_OPERATOR_TIMEOUT") + # Capture output only in memory: private SSH/CLI errors must never enter evidence. + try: + result = subprocess.run(command, capture_output=True, text=True, timeout=timeout) + except (OSError, subprocess.TimeoutExpired): + raise ContractError("OPERATOR_" + name.upper() + "_UNAVAILABLE") from None + if result.returncode: + raise ContractError("OPERATOR_" + name.upper() + "_FAILED") + try: + value = json.loads(result.stdout) + if not isinstance(value, dict): + raise ValueError() + return value + except ValueError: + raise ContractError("OPERATOR_" + name.upper() + "_INVALID_RECEIPT") from None + + +def execute(directory, plan): + """Prevent two controllers from restoring or activating the same run concurrently.""" + import fcntl + with (directory / "execute.lock").open("a+") as lock: + os.fchmod(lock.fileno(), 0o600) + try: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + raise ContractError("RUN_ALREADY_LOCKED") from None + return _execute(directory, plan) + + +def _execute(directory, plan): + old = events(directory) + cutoffs = [e for e in old if e["kind"] == "cutoff"] + if len(cutoffs) != 1 or any(e["kind"] == "restore_started" for e in old): + raise ContractError("RUN_NOT_READY_OR_ALREADY_STARTED") + if digest(load(directory / "inputs.json")) != cutoffs[0]["data"]["inputs_sha256"]: + raise ContractError("FROZEN_INPUTS_CHANGED") + # 1. Validate every operator and cleanup command before any remote action. + if set(plan) != {"restore", "snapshot", "activate", "functional", "final_snapshot", "cleanup"}: + raise ContractError("INCOMPLETE_OPERATOR_PLAN") + for action in plan.values(): + if (not isinstance(action, dict) or not isinstance(action.get("argv"), list) + or not action["argv"] or not all(isinstance(x, str) for x in action["argv"]) + or type(action.get("timeout_seconds", 300)) is not int + or not 1 <= action.get("timeout_seconds", 300) <= 1800): + raise ContractError("INVALID_OPERATOR_COMMAND") + t0, failure, functional, cleanup, audit = cutoffs[0]["at_utc"], None, {}, {}, None + final_audit = None + append(directory, "restore_started", {}) + started = time.monotonic() + try: + # 2. Restore only the frozen inputs; a receipt binds their digest. + receipt = execute_operator(plan, "restore") + if (receipt.get("restored") is not True or receipt.get("source_access_used") is not False + or receipt.get("inputs_sha256") != cutoffs[0]["data"]["inputs_sha256"]): + raise ContractError("RESTORE_SOURCE_DEPENDENCY") + append(directory, "restored", {"elapsed_seconds": round(time.monotonic() - started, 3)}) + # 3. Persist the pristine DB/GCS audit before any application mutation. + audit = initial_audit(directory, execute_operator(plan, "snapshot")) + if not audit["activation_allowed"]: + raise ContractError("INITIAL_AUDIT_BLOCKS_ACTIVATION") + # 4. Only a durable, passing initial gate permits bucket rebinding and Worker activation. + if execute_operator(plan, "activate").get("activated") is not True: + raise ContractError("ACTIVATION_FAILED") + activated = append(directory, "activated", {}) + # 5. Keep each actual feature observation and its recovery time separate. + observations = execute_operator(plan, "functional") + if set(observations) != set(FUNCTIONS): + raise ContractError("MISSING_FUNCTIONAL_CHECK") + for name in FUNCTIONS: + item = observations[name] + if (not isinstance(item, dict) or set(item) != {"passed", "samples", "at_utc"} + or item.get("passed") is not True or not positive(item.get("samples"))): + raise ContractError("FUNCTIONAL_" + name.upper() + "_FAILED") + at = instant(item.get("at_utc")) + if not instant(activated["at_utc"]) <= at <= datetime.now(timezone.utc): + raise ContractError("INVALID_FUNCTIONAL_TIMESTAMP") + functional[name] = {"passed": True, "samples": item["samples"], "at_utc": at.isoformat(), + "from_cutoff_seconds": round((at - instant(t0)).total_seconds(), 3)} + append(directory, "functional", {"function": name, "samples": item["samples"], + "recovered_at_utc": item["at_utc"], + "from_cutoff_seconds": round((at - instant(t0)).total_seconds(), 3)}) + # 6. Final convergence cannot replace or overwrite the initial loss audit. + final_snapshot = execute_operator(plan, "final_snapshot") + counters = final_snapshot.get("consistency", {}) + if (set(counters) != {"checked_documents", "incomplete_jobs", "duplicate_chunks", "duplicate_embeddings", "missing_embeddings"} + or not positive(counters.get("checked_documents")) or any( + type(counters.get(k)) is not int or counters[k] != 0 + for k in ("incomplete_jobs", "duplicate_chunks", "duplicate_embeddings", "missing_embeddings"))): + raise ContractError("FINAL_CONSISTENCY_FAILED") + final_audit = audit_records(requests(directory), final_snapshot, pristine=False, + originals=load(directory / "inputs.json")["originals"]) + write_private(directory / "final-snapshot.json", json_text(final_snapshot), exclusive=True) + write_private(directory / "final-audit.json", json_text(final_audit), exclusive=True) + append(directory, "final_audit", {"snapshot_sha256": digest(final_snapshot), + "audit_sha256": digest(final_audit), "consistency": counters}) + except (ContractError, OSError, KeyError, TypeError, AttributeError, KeyboardInterrupt) as error: + failure = str(error) if isinstance(error, ContractError) else ("RUN_INTERRUPTED" if isinstance(error, KeyboardInterrupt) else "INVALID_STAGE_EVIDENCE") + append(directory, "interrupted", {"reason": failure}) + finally: + # 7. Always attempt owned-resource cleanup; missing confirmation prevents PASS. + try: + cleanup = execute_operator(plan, "cleanup") + if any(cleanup.get(k) is not True for k in + ("public_config_preserved", "source_cluster_restored", "owned_resources_removed")): + raise ContractError("CLEANUP_NOT_CONFIRMED") + append(directory, "cleanup", {k: cleanup[k] for k in + ("public_config_preserved", "source_cluster_restored", "owned_resources_removed")}) + except ContractError as error: + failure = failure or str(error) + append(directory, "cleanup_unconfirmed", {"reason": str(error)}) + summary = {"run_id": load(directory / "manifest.json")["run_id"], + "execution": "PASS" if failure is None else "INCOMPLETE", + "reason": failure, "rpo_guarantee": "NOT_EVALUATED", + "scope": "isolated_document_download_search_indexing", + "initial_audit": audit, "final_audit": final_audit, + "functional_checks": functional} + replace_private(directory / "summary.json", json_text(summary)) + append(directory, "finished", {"execution": summary["execution"], "summary_sha256": digest(summary)}) + return summary + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("action", choices=["init", "sent", "response", "preflight", "freeze", "execute"]) + parser.add_argument("--run-dir", type=Path) + parser.add_argument("--run-id") + parser.add_argument("--revision") + parser.add_argument("--input", type=Path) + args = parser.parse_args(argv) + try: + if args.action == "init": + initialize(args.run_dir, args.run_id, args.revision) + result = {"initialized": True} + elif args.action == "preflight": + result = validate_inputs(load(args.input)) + elif args.action == "sent": + record_sent(args.run_dir, load(args.input)); result = {"recorded": True} + elif args.action == "response": + record_response(args.run_dir, load(args.input)); result = {"recorded": True} + elif args.action == "freeze": + freeze_inputs(args.run_dir, load(args.input)); result = {"frozen": True} + else: + summary = execute(args.run_dir, load(args.input)) + result = {"execution": summary["execution"], "reason": summary["reason"]} + print(json.dumps(result, ensure_ascii=False)) + return 0 if result.get("ready", True) and result.get("execution") != "INCOMPLETE" else 1 + except (ContractError, OSError, TypeError) as error: + print(json.dumps({"result": "NO_GO", "reason": str(error) if isinstance(error, ContractError) + else "LOCAL_INPUT_OR_EVIDENCE_ERROR"})) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/opensql/test_dr_recovery.py b/scripts/opensql/test_dr_recovery.py new file mode 100644 index 00000000..909f9b95 --- /dev/null +++ b/scripts/opensql/test_dr_recovery.py @@ -0,0 +1,300 @@ +#!/usr/bin/env python3 +"""Contract tests use synthetic JSON and local child commands, never GCP or a DB.""" +import copy +from datetime import datetime, timedelta, timezone +import json +from pathlib import Path +import sys +import tempfile +import unittest + +import dr_recovery as dr + + +class RecoveryContractTest(unittest.TestCase): + """Keep first-restoration evidence immutable before any simulated Worker activation.""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + self.run = self.root / "run" + dr.initialize(self.run, "DR-CONTRACT-01", "4cfe966") + self.inputs = {"schema_version": 1, "captured_at_utc": dr.utc_now(), + "checks": {name: True for name in dr.CHECKS}, + "backup": {"id": "20261018T010000", "sha256": "a" * 64, "size": 100, "generation": "1"}, + "identity_sha256": "b" * 64, + "wals": [{"name": "000000010000000000000001", "sha256": "c" * 64, + "size": 16777216, "generation": "2"}], + "originals": [{"fileObjectId": 3, "sha256": "d" * 64, "size": 33, "generation": "7"}]} + + def sent(self, key="r1", cohort="functional", status=201): + record = {"request_id": key, "cohort": cohort, "sha256": "d" * 64, "size": 33} + dr.record_sent(self.run, record) + ids = dict(zip(dr.IDS, (1, 2, 3, 4))) + dr.record_response(self.run, {"request_id": key, "http_status": status, + **({"ids": ids} if status in (200, 201) else {})}) + return {**record, "http_status": status, + "outcome": "ACKNOWLEDGED" if status in (200, 201) else "UNKNOWN" if status is None else "HTTP_ERROR", + **({"ids": ids} if status in (200, 201) else {})} + + def snapshot(self, states=None): + rows = [] + for r in dr.requests(self.run): + state = (states or {}).get(r["request_id"], "RESTORED") + rows.append({"request_id": r["request_id"], "state": state, + "ids": dict(zip(dr.IDS, (1, 2, 3, 4))), "relations_match": True, + "sha256": r["sha256"], "size": r["size"], "generation": "7"}) + return {"worker_started": False, "flyway_ran": False, "repair_ran": False, + "source_access_used": False, "records": rows} + + def plan(self, snapshot=None, fail=None): + snapshot = snapshot or self.snapshot() + receipts = {"restore": {"restored": True, "source_access_used": False, "inputs_sha256": dr.digest(self.inputs)}, + "snapshot": snapshot, "activate": {"activated": True}, + "cleanup": {"public_config_preserved": True, "source_cluster_restored": True, + "owned_resources_removed": True}} + final = self.snapshot() + final["worker_started"] = True + final["consistency"] = {"checked_documents": 2, "incomplete_jobs": 0, + "duplicate_chunks": 0, "duplicate_embeddings": 0, "missing_embeddings": 0} + receipts["final_snapshot"] = final + fixture = self.root / "receipts.json" + fixture.write_text(json.dumps(receipts)) + script = self.root / "controlled_driver.py" + script.write_text('''import datetime,json,pathlib,sys +root=pathlib.Path(sys.argv[1]); phase=sys.argv[2] +with (root/'calls.txt').open('a') as f:f.write(phase+'\\n') +if phase==sys.argv[3]:raise SystemExit(1) +if phase=='activate': + assert (root/'run'/'initial-audit.json').exists() + assert json.loads((root/'run'/'initial-audit.json').read_text())['activation_allowed'] +if phase=='functional': + print(json.dumps({name:{'passed':True,'samples':1,'at_utc':datetime.datetime.now(datetime.timezone.utc).isoformat()} for name in ['download','search','job_resume','new_upload','new_indexing','new_download']})) +else:print(json.dumps(json.loads((root/'receipts.json').read_text())[phase])) +''') + return {name: {"argv": [sys.executable, str(script), str(self.root), name, fail or "none"], + "timeout_seconds": 5} for name in + ("restore", "snapshot", "activate", "functional", "final_snapshot", "cleanup")} + + def ready(self): + dr.freeze_inputs(self.run, self.inputs) + + def test_inputs_missing_check_is_no_go(self): + for name in dr.CHECKS: + with self.subTest(name=name): + value = copy.deepcopy(self.inputs); value["checks"][name] = False + self.assertFalse(dr.validate_inputs(value)["ready"]) + + def test_partial_wal_rejected(self): + self.inputs["wals"][0]["name"] += ".partial" + with self.assertRaisesRegex(dr.ContractError, "INVALID_WAL_OBJECT"): + dr.validate_inputs(self.inputs) + + def test_duplicate_wal_rejected(self): + self.inputs["wals"] *= 2 + with self.assertRaisesRegex(dr.ContractError, "INVALID_WAL_OBJECT"): + dr.validate_inputs(self.inputs) + + def test_original_identity_must_be_preserved(self): + self.inputs.pop("identity_sha256") + with self.assertRaisesRegex(dr.ContractError, "NO_OFFSITE_IDENTITY"): + self.ready() + + def test_frozen_inputs_cannot_be_replaced(self): + self.ready() + with self.assertRaisesRegex(dr.ContractError, "INPUTS_ALREADY_FROZEN"): + self.ready() + + def test_changed_inputs_block_operator(self): + self.sent(); self.ready() + (self.run / "inputs.json").write_text('{}') + with self.assertRaisesRegex(dr.ContractError, "FROZEN_INPUTS_CHANGED"): + dr.execute(self.run, self.plan()) + self.assertFalse((self.root / "calls.txt").exists()) + + def test_response_ledger_is_durable_and_private(self): + self.sent() + self.assertEqual(len(dr.events(self.run)), 2) + self.assertEqual((self.run / "events.jsonl").stat().st_mode & 0o777, 0o600) + self.assertEqual(self.run.stat().st_mode & 0o777, 0o700) + + def test_corrupt_ledger_is_not_summarized(self): + self.sent() + p = self.run / "events.jsonl"; p.write_text(p.read_text().replace('"size": 33', '"size": 34')) + with self.assertRaisesRegex(dr.ContractError, "CORRUPT_EVENT_LEDGER"): + dr.requests(self.run) + + def test_duplicate_response_rejected(self): + self.sent() + with self.assertRaisesRegex(dr.ContractError, "DUPLICATE_RESPONSE"): + dr.record_response(self.run, {"request_id": "r1", "http_status": None}) + + def test_success_requires_all_four_ids(self): + dr.record_sent(self.run, {"request_id": "r1", "cohort": "loss", "sha256": "d" * 64, "size": 33}) + with self.assertRaisesRegex(dr.ContractError, "SUCCESS_WITHOUT_IDS"): + dr.record_response(self.run, {"request_id": "r1", "http_status": 201, "ids": {"documentId": 1}}) + + def test_http_error_may_be_present_in_restored_db(self): + self.sent(status=500) + result = dr.audit_records(dr.requests(self.run), self.snapshot()) + self.assertEqual(result["http_error"], 1) + self.assertEqual(result["acknowledged"], 0) + self.assertEqual(result["requests"][0]["restored_state"], "RESTORED") + + def test_missing_response_is_unknown_not_failure(self): + dr.record_sent(self.run, {"request_id": "r1", "cohort": "functional", "sha256": "d" * 64, "size": 33}) + self.assertEqual(dr.requests(self.run)[0]["outcome"], "UNKNOWN") + + def test_success_response_ids_must_match_restored_relations(self): + self.sent(); s = self.snapshot(); s["records"][0]["ids"]["embeddingJobId"] = 99 + with self.assertRaisesRegex(dr.ContractError, "RESPONSE_BINDING_MISMATCH"): + dr.audit_records(dr.requests(self.run), s) + + def test_old_and_new_success_do_not_hide_middle_loss(self): + self.sent("r1"); self.sent("r2", "loss"); self.sent("r3", "loss") + result = dr.audit_records(dr.requests(self.run), self.snapshot({"r2": "DB_MISSING"})) + self.assertEqual(result["acknowledged"], 3) + self.assertEqual(result["lost_or_incomplete_acknowledged"], 1) + self.assertTrue(result["activation_allowed"]) + + def test_private_extra_snapshot_fields_are_rejected_before_recording(self): + self.sent(); snapshot = self.snapshot(); snapshot['credential'] = 'not-a-real-secret' + with self.assertRaisesRegex(dr.ContractError, "UNSAFE_SNAPSHOT_FIELDS"): + dr.audit_records(dr.requests(self.run), snapshot) + + def test_private_extra_row_fields_are_rejected_before_recording(self): + self.sent(); snapshot = self.snapshot(); snapshot['records'][0]['endpoint'] = 'private-omitted' + with self.assertRaisesRegex(dr.ContractError, "UNSAFE_SNAPSHOT_FIELDS"): + dr.audit_records(dr.requests(self.run), snapshot) + + def test_restore_receipt_must_bind_frozen_manifest(self): + self.sent(); self.ready(); plan = self.plan() + p = self.root/'receipts.json'; receipts = json.loads(p.read_text()) + receipts['restore']['inputs_sha256'] = '0' * 64; p.write_text(json.dumps(receipts)) + result = dr.execute(self.run, plan) + self.assertEqual(result['reason'], 'RESTORE_SOURCE_DEPENDENCY') + self.assertEqual((self.root/'calls.txt').read_text().splitlines(), ['restore', 'cleanup']) + + def test_missing_pinned_originals_is_no_go(self): + self.inputs.pop('originals') + with self.assertRaisesRegex(dr.ContractError, 'NO_PINNED_ORIGINALS'): + self.ready() + + def test_wrong_preserved_generation_blocks_activation(self): + self.sent(); self.ready(); snapshot = self.snapshot() + snapshot['records'][0]['generation'] = '8' + result = dr.execute(self.run, self.plan(snapshot)) + self.assertEqual(result['reason'], 'RESTORED_OBJECT_NOT_PINNED') + self.assertNotIn('activate', (self.root/'calls.txt').read_text().splitlines()) + + def test_private_consistency_field_is_rejected(self): + self.sent(); snapshot = self.snapshot() + snapshot['consistency'] = {'credential': 'synthetic-omission'} + with self.assertRaisesRegex(dr.ContractError, 'UNSAFE_SNAPSHOT_FIELDS'): + dr.audit_records(dr.requests(self.run), snapshot) + + def test_private_value_in_missing_row_is_rejected(self): + self.sent(); snapshot = self.snapshot({'r1': 'DB_MISSING'}) + snapshot['records'][0]['ids'] = {'credential': 'synthetic-omission'} + with self.assertRaisesRegex(dr.ContractError, 'INVALID_SNAPSHOT_IDS'): + dr.audit_records(dr.requests(self.run), snapshot) + + def test_final_snapshot_cannot_use_original_cluster(self): + self.sent(); snapshot = self.snapshot(); snapshot['source_access_used'] = True + with self.assertRaisesRegex(dr.ContractError, 'SNAPSHOT_SOURCE_DEPENDENCY'): + dr.audit_records(dr.requests(self.run), snapshot, pristine=False) + + def test_invalid_functional_receipt_is_not_persisted(self): + self.sent(); self.ready(); plan = self.plan() + unsafe = {k: {'passed': 'synthetic-private-marker', 'samples': 1, 'at_utc': dr.utc_now()} for k in dr.FUNCTIONS} + plan['functional']['argv'] = [sys.executable, '-c', 'print('+repr(json.dumps(unsafe))+')'] + result = dr.execute(self.run, plan) + self.assertEqual(result['execution'], 'INCOMPLETE') + self.assertNotIn('synthetic-private-marker', (self.run/'summary.json').read_text()) + self.assertEqual(result['functional_checks'], {}) + + def test_racing_response_after_audit_is_rejected_when_replayed(self): + self.sent(); self.ready() + dr.append(self.run, 'initial_audit', {}) + dr.append(self.run, 'response', {'request_id': 'r1', 'http_status': None, 'outcome': 'UNKNOWN'}) + with self.assertRaisesRegex(dr.ContractError, 'RESPONSE_AFTER_AUDIT'): + dr.requests(self.run) + + def test_object_missing_blocks_worker(self): + self.sent(); result = dr.audit_records(dr.requests(self.run), self.snapshot({"r1": "OBJECT_MISSING"})) + self.assertFalse(result["activation_allowed"]) + + def test_real_hash_mismatch_is_not_restored(self): + self.sent(); s = self.snapshot(); s["records"][0]["sha256"] = '0' * 64 + result = dr.audit_records(dr.requests(self.run), s) + self.assertEqual(result["requests"][0]["restored_state"], "HASH_MISMATCH") + self.assertFalse(result["activation_allowed"]) + + def test_missing_observation_is_not_db_loss(self): + self.sent(); s = self.snapshot(); s["records"] = [] + with self.assertRaisesRegex(dr.ContractError, "MISSING_SNAPSHOT_OBSERVATION"): + dr.audit_records(dr.requests(self.run), s) + + def test_initial_state_rejects_started_worker_or_flyway(self): + self.sent() + for key in ["worker_started", "flyway_ran", "repair_ran", "source_access_used"]: + with self.subTest(key=key): + s = self.snapshot(); s[key] = True + with self.assertRaisesRegex(dr.ContractError, "INITIAL_SNAPSHOT_NOT_PRISTINE|SNAPSHOT_SOURCE_DEPENDENCY"): + dr.audit_records(dr.requests(self.run), s) + + def test_success_stages_audit_before_activation_then_final_audit(self): + self.sent(); self.ready(); result = dr.execute(self.run, self.plan()) + self.assertEqual(result["execution"], "PASS") + self.assertEqual((self.root / 'calls.txt').read_text().splitlines(), + ['restore', 'snapshot', 'activate', 'functional', 'final_snapshot', 'cleanup']) + kinds = [e["kind"] for e in dr.events(self.run)] + self.assertLess(kinds.index('initial_audit'), kinds.index('activated')) + self.assertLess(kinds.index('activated'), kinds.index('final_audit')) + self.assertEqual(result["rpo_guarantee"], "NOT_EVALUATED") + + def test_initial_loss_preserved_when_final_snapshot_is_repaired(self): + self.sent(); self.sent('r2', 'loss'); self.ready() + result = dr.execute(self.run, self.plan(self.snapshot({'r2': 'DB_MISSING'}))) + self.assertEqual(result['execution'], 'PASS') + self.assertEqual(result['initial_audit']['lost_or_incomplete_acknowledged'], 1) + self.assertEqual(result['final_audit']['lost_or_incomplete_acknowledged'], 0) + + def test_audit_failure_never_activates_and_cleans_up(self): + self.sent(); self.ready() + result = dr.execute(self.run, self.plan(self.snapshot({'r1': 'HASH_MISMATCH'}))) + self.assertEqual(result['execution'], 'INCOMPLETE') + self.assertEqual((self.root/'calls.txt').read_text().splitlines(), ['restore', 'snapshot', 'cleanup']) + + def test_restore_failure_cleans_up(self): + self.sent(); self.ready() + result = dr.execute(self.run, self.plan(fail='restore')) + self.assertEqual(result['reason'], 'OPERATOR_RESTORE_FAILED') + self.assertEqual((self.root/'calls.txt').read_text().splitlines(), ['restore', 'cleanup']) + + def test_missing_cleanup_blocks_all_actions(self): + self.sent(); self.ready(); plan = self.plan(); plan.pop('cleanup') + with self.assertRaisesRegex(dr.ContractError, 'INCOMPLETE_OPERATOR_PLAN'): + dr.execute(self.run, plan) + self.assertFalse((self.root/'calls.txt').exists()) + + def test_bad_later_command_blocks_restore(self): + self.sent(); self.ready(); plan = self.plan(); plan['activate']['argv'] = [33] + with self.assertRaisesRegex(dr.ContractError, 'INVALID_OPERATOR_COMMAND'): + dr.execute(self.run, plan) + self.assertFalse((self.root/'calls.txt').exists()) + + def test_cleanup_failure_is_not_pass(self): + self.sent(); self.ready(); result = dr.execute(self.run, self.plan(fail='cleanup')) + self.assertEqual(result['execution'], 'INCOMPLETE') + self.assertEqual(result['reason'], 'OPERATOR_CLEANUP_FAILED') + + def test_finished_run_cannot_be_executed_again(self): + self.sent(); self.ready(); plan = self.plan(); dr.execute(self.run, plan) + with self.assertRaisesRegex(dr.ContractError, 'RUN_NOT_READY_OR_ALREADY_STARTED'): + dr.execute(self.run, plan) + + +if __name__ == '__main__': + unittest.main()