From 53685f34fc1feb42f2c11405ee4835aae21e9592 Mon Sep 17 00:00:00 2001 From: liveforpresent Date: Tue, 27 Oct 2026 19:57:04 +0900 Subject: [PATCH] test(normalization): verify V1 result pins through publication and recovery --- README.md | 36 ++ ...NG_NORMALIZATION_RESULT_VERSION_PIN_E2E.md | 130 +++++ ...nitialize-trading-normalization-compute.py | 44 ++ .../start-trading-normalization-service.ps1 | 62 +++ ...ading-normalization-result-version-pin.ps1 | 26 + ...trading-normalization-runtime-contract.ps1 | 36 +- ..._trading_normalization_runtime_contract.py | 92 +++- scripts/trading_normalization_version_pin.py | 504 ++++++++++++++++++ 8 files changed, 904 insertions(+), 26 deletions(-) create mode 100644 docs/TRADING_NORMALIZATION_RESULT_VERSION_PIN_E2E.md create mode 100644 scripts/initialize-trading-normalization-compute.py create mode 100644 scripts/start-trading-normalization-service.ps1 create mode 100644 scripts/test-trading-normalization-result-version-pin.ps1 create mode 100644 scripts/trading_normalization_version_pin.py diff --git a/README.md b/README.md index 72ef54f..ac678b5 100644 --- a/README.md +++ b/README.md @@ -89,6 +89,42 @@ Compute API 중단으로 ACK의 일시 실패를 만들고, Revision 보존과 A inputSetHash에 포함되므로 실행 간 전체 resultHash 값은 고정되지 않는다. 매 실행마다 Context 규칙으로 실제 hash를 독립 재계산해 일치 여부를 검증한다. +## Trading Normalization result version pin E2E + +```powershell +.\scripts\test-trading-normalization-result-version-pin.ps1 -CorePort 8680 -ComputePort 8681 +``` + +기존 runtime/publication harness와 동일한 경로·환경·DB parameter를 받는다. 첫 실행 검사는 +`verify-contract-sync.ps1 -IncludeWeb`이며 DB는 loopback의 `refinvest_integration`만 허용한다. +Core 공개 합성 Binance 업로드·validation부터 실제 durable create, accepted/status, +terminal descriptor/payload, LedgerRevision 및 ACK까지 V1/V1 pin을 대조한다. +기존 V1 fixed vector, 9개 section digest, 단일 publication과 ACK 실패·재시도를 재사용한다. +loopback Integration 프록시가 ACK 요청 시 별도 DB connection에서 publication commit의 +가시성까지 확인한다. 이번 합성 작업의 ACK만 일시적으로 HTTP 503으로 거절한 뒤 해제해 +실제 Core outbox 재시도와 Revision 불변을 검증한다. 프록시는 capability URL·token·CSV body를 evidence에 저장하지 않는다. + +추가 검증은 ORCH-024(같은 key/request replay·다른 쌍 충돌), 025(3개 필드 누락, +2개 교차 쌍, 3개 미지원 값), 026(descriptor와 payload 불일치 각각)을 포함한다. +부정 응답은 해당 합성 Session으로만 주입하고 실제 서비스 코드는 변경하지 않는다. +payload fault는 transport 길이·SHA-256을 맞춰 Core의 pin gate에 도달시킨다. +실패 Session의 Revision·PERSISTED ACK 0건과 기존 Book latest 보존을 확인한다. + +ORCH-027은 새 `tnorm_legacycore*`/`tnorm_legacycompute*` schema에 실제 pre-pin migration을 +적용하고 이번 합성 publication만 복제한다. pin이 없는 Run/dispatch/Job과 옛 request fingerprint를 +준비한 뒤 실제 Core V9·Compute 0017 SQL을 실행한다. 이후 두 API를 해당 schema로 재시작해 +명시적 V1/V1 HTTP status·replay와 공개 validation replay, 기존 Revision·ACK 불변을 확인한다. +실행 중인 원본 schema의 컬럼이나 trigger를 제거하지 않으며 schema·로그는 조사용으로 보존한다. + +결과는 `.integration-data/trading-normalization//evidence.json`의 `versionPin`에 +시나리오별 실행 상태와 비민감 Run/Job/Revision/ACK ID로 남긴다. 재시작 전 로그는 같은 실행 +디렉터리의 `before-migration-logs/`에 보존한다. ORCH-028은 Core의 admission/dispatch 전용 +중지 기능이 없으므로 `NOT_RUN_MISSING_CONTROL`로 기록한다. 이 항목이 있는 실행의 +스크립트 성공은 **실행 가능한 V1 pin 검증**의 성공이며 공동 전환 전체의 PASS가 아니다. +V2 Core publication은 이번 범위에서 제외한다. +정확한 assertion, 공동 전환에 필요한 후속 기능과 V2 선행 구현 상태는 +[결과 버전 pin E2E 문서](docs/TRADING_NORMALIZATION_RESULT_VERSION_PIN_E2E.md)를 참조한다. + ## Performance fixture The deterministic ten-year, five-asset synthetic fixture, six benchmark scenarios, expected-result diff --git a/docs/TRADING_NORMALIZATION_RESULT_VERSION_PIN_E2E.md b/docs/TRADING_NORMALIZATION_RESULT_VERSION_PIN_E2E.md new file mode 100644 index 0000000..9e194ae --- /dev/null +++ b/docs/TRADING_NORMALIZATION_RESULT_VERSION_PIN_E2E.md @@ -0,0 +1,130 @@ +# Trading Normalization Result Version Pin E2E + +이 harness는 Context ADR-065/066/070, `TRADING_VERSIONING.md`의 공동 전환 규칙과 +`TR-ACC-IMPORT-ORCH-024`~`028`을 추적한다. 정본·OpenAPI·제품 business code를 변경하지 않는다. + +## 실행과 소유 범위 + +```powershell +.\scripts\test-trading-normalization-result-version-pin.ps1 -CorePort 8680 -ComputePort 8681 +``` + +기존 runtime/publication script와 같은 sibling 경로 parameter를 받는다. +`verify-contract-sync.ps1 -IncludeWeb`을 프로세스·DB 변경 전에 실행한다. +loopback PostgreSQL의 `refinvest_integration`만 허용하며 Core는 `public`, Compute는 실행별 +`tnorm_*` schema를 사용한다. V1 migration 복구 검증은 별도의 새 Core/Compute schema에서 수행한다. +비민감 합성 Binance fixture만 업로드하고 정상 JWT·CSRF와 실제 HTTP/DB를 사용한다. + +기존 publication 시나리오의 transport 검증, 독립 hash 계산, 9개 section digest, +발행 관계 행·staging 검증, ACK 실패·재시도 및 중복 방지를 재사용한다. +새 loopback 프록시는 실제 create와 accepted를 관찰하고 ACK 요청마다 별도 DB connection으로 +발행 commit의 가시성을 확인한다. V2 단위 검증을 V2 Core publication으로 해석하지 않는다. +ACK 실패는 해당 합성 Job의 요청에만 프록시가 503을 반환해 만든다. 실패 outbox와 단일 Revision을 +확인한 뒤 fault를 해제하고 실제 재시도·Compute ACK 저장을 검사한다. 프로세스 중지 지연 동안 +ACK가 먼저 성공할 수 있는 기존 테스트의 타이밍 경쟁을 피한다. + +## 시나리오와 검증 근거 + +| 시나리오 | assertion | +|---|---| +| V1 pin 정상 발행 | Core Run·durable dispatch, 실제 create, Job 컬럼·request snapshot, accepted/status, descriptor/payload/digest manifest, Revision이 모두 V1/V1; input identity에 결과 버전 없음; 9개 V1 section digest; ACK 요청 전에 commit된 Revision 확인 | +| ORCH-024 | 동일 key/request는 최초 accepted와 동일한 202/Job/pin; V2/V2 요청으로 같은 key 사용 시 409 `IDEMPOTENCY_PAYLOAD_CONFLICT`; Job 전체·grant revision 불변 | +| ORCH-025 | 두 필드 누락·각 한 필드 누락은 422 `REQUEST_INVALID`; 두 교차 쌍은 422 `RESULT_VERSION_PAIR_INVALID`; schema/hash/양쪽 미지원은 422 `RESULT_SCHEMA_UNSUPPORTED`; 각 요청 뒤 Job·idempotency 수 불변 | +| ORCH-026 descriptor | 실제 Compute V1 성공 결과의 descriptor만 V2/V2로 주입; Core `PROCESSING_FAILED`/Run·dispatch `TERMINAL_FAILURE`, `RESULT_VERSION_PIN_MISMATCH`/`WAIT_FOR_RECOVERY`; Revision·ACK 0, 기존 latest 보존 | +| ORCH-026 payload | descriptor pin은 V1/V1로 유지하고 payload pin만 V2/V2로 변경; descriptor 길이·transport SHA-256은 변경 bytes와 일치시켜 pin 검증에 도달; 같은 차단 결과와 ACK 미수신 확인 | +| ORCH-027 | 새 pre-pin schema에 이번 합성 publication만 복제; pin 없는 저장 Run/dispatch/Job, 이전 logical key·fingerprint로 시작; 실제 Core V9·Compute 0017 SQL 적용; 저장 pin·request/idempotency fingerprint 일치; 양 API 재시작 후 V1 status/terminal/replay 및 public validation replay; 같은 Run·Job·Revision·ACK와 기존 hash 보존 | +| ORCH-028 | **미실행: 필요한 Core/운영 제어 부재. 공동 전환 PASS가 아님** | + +ORCH-026의 fault는 새 합성 Session ID에 한정하며 status/Job pin은 실제 V1/V1을 유지한다. +원본 Job/Run pin을 수정하거나 서비스 코드에 test flag를 넣지 않는다. payload fault는 pin gate의 +거절을 검증하기 위한 응답 변형이며 유효한 V2 산출물이나 V2 publication 근거가 아니다. + +ORCH-027은 **저장 완료된 V1 publication**의 합성 pre-pin 복구 검증이다. 실행 중인 원본 schema를 +downgrade하지 않고, 새 schema에 실제 pre-pin migration을 적용한 뒤 fixture를 채운다. +미결 create·pending 작업의 전체 fleet drain과 운영 중지/재개는 ORCH-028의 미검증 범위로 남긴다. + +Compute 0017은 기존 `initialize-compute-schema.ps1`과 같은 `psql -f`의 statement별 +autocommit 경계로 실행한다. 저장된 terminal Job이 있을 때 0017 전체를 한 transaction으로 +감싸면 `0010_create_trading_review_normalization_jobs.sql:121`의 deferred Job→terminal FK 때문에 +`0017_pin_trading_normalization_result_pair.sql:80`의 `ALTER TABLE trading_review_jobs`가 +SQLSTATE `55006`(`ObjectInUse`)으로 실패하는 것을 별도로 재현했다. SQL을 수정하거나 constraint를 +비활성화하지 않았다. 원자적 migration runner를 사용하는 배포는 Compute 소유 SQL/runner 후속 검증이 +필요하며, 이 E2E의 migration PASS를 해당 배포 방식의 PASS로 해석하지 않는다. + +## 공동 전환 후속 작업 + +Core main `0ee1053449280237dffb2fdb26124bd5a974fcc5` 기준: + +- `TradingNormalizationService.kt:33,37,68`의 validation/retry 경로에는 전환 중 admission을 + `503 TEMPORARY_DEPENDENCY_FAILURE`로 거절하는 제어가 없다. public idempotency claim 이전에 + 중지 여부를 확인하고 Session/Run/dispatch/idempotency를 바꾸지 않아야 한다. +- `TradingNormalizationDispatchScheduler.kt:10`의 `refinvest.compute.enabled`는 dispatch뿐 아니라 + poll/result/ACK도 함께 제어한다. 조회·terminal 처리·삭제·취소·복구를 계속하면서 새 create만 + 중지하는 경계로 사용할 수 없다. dispatcher의 실제 claim/send 경계에도 fleet 공통 차단이 필요하다. +- 운영 제어는 이전 Core 인스턴스와 미전송 무필드 create가 0임을 확인하고, 응답이 불명확한 create를 + Job ID·idempotency record로 대조해야 한다. 확인 불가능한 요청은 재전송하지 말고 보류/복구해야 한다. +- 중지 상태에서 V1 작업 drain 또는 provenance 기반 backfill 뒤 명시적 V1/V1 create, + accepted/status/terminal/replay를 격리 HTTP/DB에서 검증하고 성공한 뒤에만 admission·dispatch를 + 재개해야 한다. 검증 실패 시 중지를 유지하는 rollback/runbook과 권한·감사 근거가 필요하다. +- 이후 ORCH-028 E2E에는 validation·retry 각각 503과 상태/row 불변, 새 dispatch/create 0, + 기존 terminal·삭제·취소 처리 지속, 실패 시 중지 유지, 성공 뒤 같은 public key 재사용을 추가해야 한다. + +## V2 선행 구현 상태 + +Compute main `0701c4ac07ac89281fa6f1b263bde294b0db6452`에는 V1/V2 쌍과 V2 producer·보충 근거 +경로가 있다. Core main에는 V2 streaming/staged verifier, `trading_ledger_analytics_supplement` +불변 저장 migration V8 및 Analytics input builder가 존재한다. 그러나 +`TradingNormalizationService.kt:138`은 신규 admission을 V1/V1로 선택한다. +이번 검증은 이를 변경하지 않으며, 명시적 V2 선택 활성화 및 실제 보충 근거 검증·불변 저장·Core +발행 E2E는 별도 전환과 검증이 필요하다. 이 문서는 그 경로의 runtime PASS를 주장하지 않는다. + +## 증적과 실패 조사 + +실행별 `.integration-data/trading-normalization//evidence.json`에 checkout commit, +시나리오별 PASS/FAIL/NOT_RUN, 비민감 Session/Run/Job/Revision/ACK ID, count·hash와 로그 경로를 남긴다. +서명 locator·credential·CSV·canonical record 본문을 evidence에 기록하지 않는다. +재시작 전 로그는 `before-migration-logs/`, 기동 시도 기록은 `*-startup.json`과 보존된 로그에 둔다. +실패에는 Core Session/Run/dispatch/staging/Revision/ACK와 Compute Job의 안전한 진단을 덧붙인다. + +Windows의 알려진 `10014 Bad address` 연결/기동 오류만 최대 3회 재시도한다. DB mutation은 +자동 재전송하지 않으며 Compute 초기 migration은 한 연결에서 파일별 transaction과 적용 이력을 +함께 commit한다. 다른 startup 오류는 즉시 실패한다. 실패한 실행은 최종 성공 실행과 별도로 보존한다. +로컬 원본 evidence·DB·환경 파일은 커밋하지 않는다. + +## 2026-10-27 실행 결과 + +Context `06fdd057`, Core `0ee10534`, Compute `0701c4ac`, Web `2ef51669`의 원격 main과 +checkout 일치를 확인했다. 위 명령은 270.687초에 종료 코드 0으로 완료됐다. +아래 PASS는 실행 가능한 V1 검증의 결과다. 공동 전환 전체의 완료를 뜻하지 않는다. + +| 실행 항목 | 결과 | +|---|---| +| 계약·문서·Web snapshot 및 V1/V2 vector preflight | PASS | +| V1/V1 pin 정상 발행·commit 뒤 ACK·실패 재시도 | PASS | +| ORCH-024 / ORCH-025 | PASS / PASS: 8개 invalid create 각각 검증 | +| ORCH-026 | PASS: descriptor와 payload 각각 실패 처리·Revision/ACK 0 | +| ORCH-027 | PASS: 기존 psql runner, 양 API 재시작·HTTP replay·저장 불변 | +| 기존 V1 result integrity / publication 회귀 | PASS: 동일 verifier·publication 경로 재사용, 9개 tamper 검사 포함 | +| ORCH-028 | NOT_RUN_MISSING_CONTROL | +| V2 Core 발행 E2E | 이번 완료 범위에서 제외 | +| 추가 atomic Compute 0017 실행 | FAIL: SQLSTATE 55006; 위 runner 제약과 후속 확인 참조 | + +정상 Run은 `0f033941-e41c-4a91-a854-6d1b85ca6b89`, Job은 +`98781090-529a-446b-ac30-7eb161706638`, Revision은 +`f29b9a74-443c-41c1-80d2-fa0b1afe99e2`, ACK outbox는 +`baf1c0de-28ce-4948-afc1-d918af52ae77`이다. ACK의 실패·성공 요청 모두 별도 연결에서 +발행 commit을 확인했고 Revision 1, canonical record 13, Evidence 5, ReviewUnit 1, +allocation 4, staging 0을 확인했다. + +정상 실행의 원본 증적은 `.integration-data/trading-normalization/8b3ce4db1e074d658008245c94ce2d14/`의 +`evidence.json`, `post-run-db-verification.json`, `before-migration-logs/`, `migration-startup/`에 있다. +descriptor fault Run/Job은 `7c955680-381f-4327-99b2-23deea062a78` / +`c6c20458-ba3e-486d-bf24-547635b6a11c`, payload fault Run/Job은 +`35828726-6db3-4345-8085-9303f0f27ada` / `3f149906-ec15-49fa-aa2f-ccf8329ad3e1`이다. +두 경우 기존 latest Revision을 보존했고 PERSISTED ACK는 없다. + +atomic runner 실패 원본은 `.integration-data/trading-normalization/58453b2292f54a158abae2f110abb515/evidence.json`과 +`tnorm_legacycomputef593dff7dd8f` schema에 보존했다. 같은 dedicated DB의 +새 `tnorm_legacycore290092f3ebfd` / `tnorm_legacycompute290092f3ebfd`에서 기존 runner의 +정상 복구와 재시작을 검증했다. 초기 환경·harness 실패는 `.integration-data/pin-harness-failed-attempts.json`에 +별도로 기록했으며 성공으로 재분류하지 않았다. diff --git a/scripts/initialize-trading-normalization-compute.py b/scripts/initialize-trading-normalization-compute.py new file mode 100644 index 0000000..8b8b588 --- /dev/null +++ b/scripts/initialize-trading-normalization-compute.py @@ -0,0 +1,44 @@ +"""Apply the real Compute migrations atomically over one integration-only connection.""" +import argparse +import os +from pathlib import Path +import time + +import psycopg +from psycopg import sql +from psycopg.conninfo import conninfo_to_dict + +parser = argparse.ArgumentParser() +parser.add_argument("--compute-path", type=Path, required=True) +parser.add_argument("--schema", required=True) +args = parser.parse_args() +settings = conninfo_to_dict(os.environ["REFINVEST_DATABASE_URL"]) +if settings.get("dbname") != "refinvest_integration" or settings.get("host") not in {"localhost", "127.0.0.1", "::1"}: + raise SystemExit("Integration-only database required") +if not args.schema.startswith("tnorm_") or not args.schema[6:].isalnum(): + raise SystemExit("Invalid runtime schema") +for attempt in range(3): + try: + connection = psycopg.connect(**settings, autocommit=True) + break + except psycopg.OperationalError as error: + if attempt == 2 or not any(code in str(error) for code in ("Bad address", "10014")): + raise SystemExit("Integration database connection failed") from None + time.sleep(0.5) +with connection: + if connection.execute("select current_database()").fetchone()[0] != "refinvest_integration": + raise SystemExit("Integration database identity mismatch") + connection.execute(sql.SQL("CREATE SCHEMA IF NOT EXISTS {}").format(sql.Identifier(args.schema))) + connection.execute(sql.SQL("SET search_path TO {}").format(sql.Identifier(args.schema))) + connection.execute("CREATE TABLE IF NOT EXISTS integration_schema_migrations (filename TEXT PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP)") + migrations = sorted((args.compute_path / "migrations").glob("*.sql")) + if not migrations: + raise SystemExit("Compute migrations unavailable") + for path in migrations: + if connection.execute("select 1 from integration_schema_migrations where filename=%s", (path.name,)).fetchone(): + continue + with connection.transaction(): + connection.execute(path.read_text(encoding="utf-8"), prepare=False) + connection.execute("insert into integration_schema_migrations(filename) values (%s)", (path.name,)) + print("APPLY " + path.name) +print("Compute migrations ready in refinvest_integration/" + args.schema) diff --git a/scripts/start-trading-normalization-service.ps1 b/scripts/start-trading-normalization-service.ps1 new file mode 100644 index 0000000..73e40a9 --- /dev/null +++ b/scripts/start-trading-normalization-service.ps1 @@ -0,0 +1,62 @@ +[CmdletBinding()] +param( + [Parameter(Mandatory)][string]$EnvironmentFile, + [string]$CorePath, + [string]$ComputePath, + [string]$ComputeBaseUrl, + [ValidateSet('core-api','compute-api')][string]$Service = 'core-api', + [Parameter(Mandatory)][string]$RegistryId, + [Parameter(Mandatory)][string]$EvidenceDirectory, + [string]$DatabaseSchema +) +$ErrorActionPreference = 'Stop' +Import-Module (Join-Path $PSScriptRoot 'IntegrationEnvironment.psm1') -Force +$values = Get-IntegrationEnvironment $EnvironmentFile +if ($Service -eq 'core-api') { + $uri = "http://127.0.0.1:$($values.INTEGRATION_CORE_PORT)/actuator/health" + $arguments = @{ EnvironmentFile=$EnvironmentFile; CoreApiPath=$CorePath; ComputeBaseUrl=$ComputeBaseUrl + RegistryId=$RegistryId; Profile=$RegistryId } + if ($DatabaseSchema) { $arguments.DatabaseSchema = $DatabaseSchema } + $startScript = 'start-core-api.ps1' +} else { + $uri = "http://127.0.0.1:$($values.INTEGRATION_COMPUTE_PORT)/health" + $arguments = @{ EnvironmentFile=$EnvironmentFile; ComputeApiPath=$ComputePath; Schema=$DatabaseSchema + RegistryId=$RegistryId; Profile=$RegistryId } + $startScript = 'start-compute-api.ps1' +} +$logRoot = Join-Path $PSScriptRoot '../.integration-data/logs' +$observations = @() +for ($attempt = 1; $attempt -le 3; $attempt++) { + $started = & (Join-Path $PSScriptRoot $startScript) @arguments + $deadline = (Get-Date).AddSeconds(600) + $ready = $false + do { + try { + $response = Invoke-WebRequest -Uri $uri -UseBasicParsing -TimeoutSec 3 + $ready = $response.StatusCode -eq 200 + } catch { } + if ($ready -or -not (Get-Process -Id $started.processId -ErrorAction SilentlyContinue)) { break } + Start-Sleep -Seconds 2 + } while ((Get-Date) -lt $deadline) + if ($ready) { + $observations += @{ attempt=$attempt; result='READY' } + $observations | ConvertTo-Json | Set-Content -LiteralPath (Join-Path $EvidenceDirectory "$Service-startup.json") -Encoding UTF8 + Write-Host "Ready: $uri (200)" + return + } + $stdout = Join-Path $logRoot "$RegistryId-$Service.out.log" + $stderr = Join-Path $logRoot "$RegistryId-$Service.err.log" + $logText = @($stdout,$stderr) | Where-Object { Test-Path -LiteralPath $_ } | ForEach-Object { Get-Content -LiteralPath $_ -Raw } + $knownSocketFailure = ($logText -join "`n") -match 'java.net.SocketException: Bad address: listen|WinError 10014' + $observations += @{ attempt=$attempt; result=if ($knownSocketFailure) { 'WINDOWS_SOCKET_LISTEN_FAILED' } else { 'STARTUP_FAILED' } } + foreach ($suffix in @('out','err')) { + $source = Join-Path $logRoot "$RegistryId-$Service.$suffix.log" + if (Test-Path -LiteralPath $source) { + Copy-Item -LiteralPath $source -Destination (Join-Path $EvidenceDirectory "$Service-startup-$attempt.$suffix.log") + } + } + $observations | ConvertTo-Json | Set-Content -LiteralPath (Join-Path $EvidenceDirectory "$Service-startup.json") -Encoding UTF8 + & (Join-Path $PSScriptRoot 'stop-integration.ps1') -RegistryId $RegistryId -Service $Service | Out-Null + if (-not $knownSocketFailure -or $attempt -eq 3) { throw "Owned $Service startup failed; see startup evidence and preserved logs." } + Write-Host "Retrying owned $Service startup after Windows socket listen failure ($attempt/3)." +} diff --git a/scripts/test-trading-normalization-result-version-pin.ps1 b/scripts/test-trading-normalization-result-version-pin.ps1 new file mode 100644 index 0000000..51e358e --- /dev/null +++ b/scripts/test-trading-normalization-result-version-pin.ps1 @@ -0,0 +1,26 @@ +[CmdletBinding()] +param( + [string]$TestRoot, + [string]$ContextPath, + [string]$CorePath, + [string]$ComputePath, + [string]$WebPath, + [string]$EnvironmentFile, + [int]$CorePort, + [int]$ComputePort, + [string]$DatabaseName = 'refinvest_integration', + [int]$TimeoutSeconds = 180 +) +$ErrorActionPreference = 'Stop' +try { + $arguments = @{ DatabaseName = $DatabaseName; TimeoutSeconds = $TimeoutSeconds; ResultVersionPin = $true } + foreach ($name in @('TestRoot', 'ContextPath', 'CorePath', 'ComputePath', 'WebPath', + 'EnvironmentFile', 'CorePort', 'ComputePort')) { + if (Get-Variable -Name $name -ValueOnly) { $arguments[$name] = Get-Variable -Name $name -ValueOnly } + } + & (Join-Path $PSScriptRoot 'test-trading-normalization-runtime-contract.ps1') @arguments + if (-not $?) { throw 'Result version pin runtime failed.' } +} catch { + [Console]::Error.WriteLine("Result version pin E2E failed: $($_.Exception.Message)") + exit 1 +} diff --git a/scripts/test-trading-normalization-runtime-contract.ps1 b/scripts/test-trading-normalization-runtime-contract.ps1 index 7ae10b7..e930b43 100644 --- a/scripts/test-trading-normalization-runtime-contract.ps1 +++ b/scripts/test-trading-normalization-runtime-contract.ps1 @@ -11,9 +11,11 @@ param( [string]$DatabaseName = 'refinvest_integration', [int]$TimeoutSeconds = 180, [switch]$ResultIntegrity, - [switch]$Publication + [switch]$Publication, + [switch]$ResultVersionPin ) $ErrorActionPreference = 'Stop' +if ($ResultVersionPin) { $Publication = $true } if (-not $TestRoot) { $TestRoot = Join-Path $PSScriptRoot '..' } if (-not $ContextPath) { $ContextPath = Join-Path $TestRoot '..\context' } if (-not $CorePath) { $CorePath = Join-Path $TestRoot '..\api\refinvest' } @@ -163,16 +165,31 @@ try { -CorePath $CorePath if ($LASTEXITCODE -ne 0) { throw 'Core trading migration failed.' } } else { Write-Host 'Core trading migrations are already applied in the integration database.' } - & (Join-Path $PSScriptRoot 'initialize-compute-schema.ps1') -EnvironmentFile $activeEnvironmentFile ` - -ComposeFile (Join-Path $TestRoot 'compose.yaml') -ComputeApiPath $ComputePath -Schema $computeSchema + if ($ResultVersionPin) { + Set-ComputeIntegrationEnvironment $values $TestRoot $computeSchema + & $python -B (Join-Path $PSScriptRoot 'initialize-trading-normalization-compute.py') ` + --compute-path $ComputePath --schema $computeSchema + } else { + & (Join-Path $PSScriptRoot 'initialize-compute-schema.ps1') -EnvironmentFile $activeEnvironmentFile ` + -ComposeFile (Join-Path $TestRoot 'compose.yaml') -ComputeApiPath $ComputePath -Schema $computeSchema + } if ($LASTEXITCODE -ne 0) { throw 'Compute migration failed.' } Remove-Item Env:PGOPTIONS -ErrorAction SilentlyContinue $scenario = 'startup' $started = $true - & (Join-Path $PSScriptRoot 'start-compute-api.ps1') -EnvironmentFile $activeEnvironmentFile ` - -ComputeApiPath $ComputePath -Schema $computeSchema -Profile $registry -RegistryId $registry - & (Join-Path $PSScriptRoot 'start-core-api.ps1') -EnvironmentFile $activeEnvironmentFile ` - -CoreApiPath $CorePath -Profile $registry -RegistryId $registry + if ($ResultVersionPin) { + & (Join-Path $PSScriptRoot 'start-trading-normalization-service.ps1') -EnvironmentFile $activeEnvironmentFile ` + -Service compute-api -ComputePath $ComputePath -DatabaseSchema $computeSchema ` + -RegistryId $registry -EvidenceDirectory $runDir + & (Join-Path $PSScriptRoot 'start-trading-normalization-service.ps1') -EnvironmentFile $activeEnvironmentFile ` + -CorePath $CorePath -RegistryId $registry -ComputeBaseUrl "http://127.0.0.1:$fixturePort" ` + -EvidenceDirectory $runDir + } else { + & (Join-Path $PSScriptRoot 'start-compute-api.ps1') -EnvironmentFile $activeEnvironmentFile ` + -ComputeApiPath $ComputePath -Schema $computeSchema -Profile $registry -RegistryId $registry + & (Join-Path $PSScriptRoot 'start-core-api.ps1') -EnvironmentFile $activeEnvironmentFile ` + -CoreApiPath $CorePath -Profile $registry -RegistryId $registry + } & (Join-Path $PSScriptRoot 'wait-for-http.ps1') -Uri "http://127.0.0.1:$ComputePort/health" -TimeoutSeconds 90 & (Join-Path $PSScriptRoot 'wait-for-http.ps1') -Uri "http://127.0.0.1:$CorePort/actuator/health" -TimeoutSeconds 600 # Start the worker only after API health and a fresh DB connection. On Windows, @@ -209,12 +226,14 @@ try { $env:TRADING_RUNTIME_DB_NAME = $DatabaseName $env:TRADING_RUNTIME_DB_USER = $values.POSTGRES_USER $env:TRADING_RUNTIME_DB_PASSWORD = $values.POSTGRES_PASSWORD + $env:TRADING_RUNTIME_PSQL = $psql $env:TRADING_RUNTIME_COMPUTE_SCHEMA = $computeSchema $env:TRADING_RUNTIME_API_KEY = $values.REFINVEST_INTERNAL_API_KEY $env:TRADING_RUNTIME_VERSION_SET = $versionJson $runtimeArguments = @() if ($ResultIntegrity) { $runtimeArguments += '--result-integrity' } if ($Publication) { $runtimeArguments += '--publication' } + if ($ResultVersionPin) { $runtimeArguments += '--result-version-pin' } & $python (Join-Path $PSScriptRoot 'test_trading_normalization_runtime_contract.py') ` --core-url "http://127.0.0.1:$CorePort" --compute-url "http://127.0.0.1:$ComputePort" ` --fixture-root $fixtureRoot --context-path $ContextPath --run-dir $runDir ` @@ -222,6 +241,9 @@ try { --core-path $CorePath --compute-path $ComputePath --compute-schema $computeSchema --registry $registry ` --timeout $TimeoutSeconds @runtimeArguments if ($LASTEXITCODE -ne 0) { throw 'Runtime contract scenario failed. See evidence and logs.' } + if ($ResultVersionPin) { + Write-Host 'V1 result pin executable scenarios PASS; ORCH-028 NOT RUN (missing admission/dispatch pause control).' + } Write-Host "Trading Normalization runtime contract PASS. Evidence: $runDir" } catch { [Console]::Error.WriteLine("Scenario $scenario failed: $($_.Exception.Message)") diff --git a/scripts/test_trading_normalization_runtime_contract.py b/scripts/test_trading_normalization_runtime_contract.py index 6a368ca..61d08c4 100644 --- a/scripts/test_trading_normalization_runtime_contract.py +++ b/scripts/test_trading_normalization_runtime_contract.py @@ -85,15 +85,23 @@ def db_connection(schema: str) -> psycopg.Connection: require(name == "refinvest_integration", "UNSAFE_DATABASE_NAME") require(schema == "public" or (schema.startswith("tnorm_") and schema[6:].isalnum()), "UNSAFE_DATABASE_SCHEMA") - connection = psycopg.connect( - host=os.environ["TRADING_RUNTIME_DB_HOST"], - port=int(os.environ["TRADING_RUNTIME_DB_PORT"]), - dbname=name, - user=os.environ["TRADING_RUNTIME_DB_USER"], - password=os.environ["TRADING_RUNTIME_DB_PASSWORD"], - options=f"-c search_path={schema}", - autocommit=True, - ) + for attempt in range(3): + try: + connection = psycopg.connect( + host=os.environ["TRADING_RUNTIME_DB_HOST"], + port=int(os.environ["TRADING_RUNTIME_DB_PORT"]), + dbname=name, + user=os.environ["TRADING_RUNTIME_DB_USER"], + password=os.environ["TRADING_RUNTIME_DB_PASSWORD"], + options=f"-c search_path={schema}", + autocommit=True, + ) + break + except psycopg.OperationalError as error: + # Retry connection establishment only; never replay a DB mutation. + if attempt == 2 or not any(code in str(error) for code in ("Bad address", "10014")): + raise + time.sleep(0.5) with connection.cursor() as cursor: cursor.execute("select current_database(), current_schema()") require(cursor.fetchone() == (name, schema), "UNSAFE_DATABASE_TARGET") @@ -273,7 +281,8 @@ def verify_publication(args, evidence: dict, core: httpx.Client, compute: httpx. connection: psycopg.Connection, compute_db: psycopg.Connection, book: dict, session_id: str, run_id: str, job_id: str, run: dict, job: dict, status: dict, descriptor: dict, - payload: dict, result: dict, uploaded: dict, artifacts: dict) -> None: + payload: dict, result: dict, uploaded: dict, artifacts: dict, + pin_proxy=None) -> None: require(payload["attemptToken"] == descriptor["attemptToken"] == status["attemptToken"], "TERMINAL_ATTEMPT_DRIFT") attempt = row(connection, """select attempt_number,attempt_token::text,compute_job_id @@ -412,20 +421,25 @@ def verify_publication(args, evidence: dict, core: httpx.Client, compute: httpx. "identity": "PASS", "evidenceReferences": "PASS"} require(ack["status"] == "PENDING", "ACK_FAULT_WINDOW_MISSED") scripts = args.test_root / "scripts" - process_script(scripts / "stop-integration.ps1", "-RegistryId", args.registry, - "-Service", "compute-api") + if pin_proxy is None: + process_script(scripts / "stop-integration.ps1", "-RegistryId", args.registry, + "-Service", "compute-api") failed_ack = poll(lambda: row(connection, """select status,attempt_count,last_failure_code from trading_normalization_ack_dispatches where run_id=%s::uuid""", (run_id,)), lambda a: a["attempt_count"] > 0 and a["last_failure_code"] is not None, - args.timeout, "ACK failure while Compute API stopped") + args.timeout, "ACK failure under controlled HTTP fault") require(failed_ack["status"] == "PENDING" and failed_ack["last_failure_code"] == "TEMPORARY_DEPENDENCY_FAILURE" and publication_counts(connection, revision_id, book["id"], run_id) == counts, "ACK_FAILURE_CHANGED_PUBLICATION") - process_script(scripts / "start-compute-api.ps1", "-EnvironmentFile", - str(args.environment_file), "-ComputeApiPath", str(args.compute_path), - "-Schema", args.compute_schema, "-Profile", args.registry, - "-RegistryId", args.registry) + if pin_proxy is not None: + require(pin_proxy.injected_ack_failures > 0, "ACK_FAULT_NOT_OBSERVED") + pin_proxy.reject_acks = False + else: + process_script(scripts / "start-compute-api.ps1", "-EnvironmentFile", + str(args.environment_file), "-ComputeApiPath", str(args.compute_path), + "-Schema", args.compute_schema, "-Profile", args.registry, + "-RegistryId", args.registry) poll(lambda: health_code(compute, f"{args.compute_url}/health"), lambda code: code == 200, args.timeout, "Compute API restart") completed_ack = poll(lambda: row(connection, """select status,attempt_count @@ -481,7 +495,10 @@ def main() -> int: parser.add_argument("--timeout", type=int, default=180) parser.add_argument("--result-integrity", action="store_true") parser.add_argument("--publication", action="store_true") + parser.add_argument("--result-version-pin", action="store_true") args = parser.parse_args() + if args.result_version_pin: + args.publication = True if args.result_integrity or args.publication: self_test(args.context_path) evidence: dict[str, object] = { @@ -496,6 +513,14 @@ def main() -> int: evidence["resultFixedVector"] = "PASS" evidence["runtimeIdentityGate"] = os.environ.get("TRADING_RUNTIME_IDENTITY_GATE", "NOT_RUN") evidence["resultHashSpecificationVersion"] = SPEC + if args.result_version_pin: + evidence["versionPin"] = {"scenarios": {name: {"result": "NOT_RUN"} for name in ( + "V1_PIN_PUBLICATION", "TR-ACC-IMPORT-ORCH-024", "TR-ACC-IMPORT-ORCH-025", + "TR-ACC-IMPORT-ORCH-026", "TR-ACC-IMPORT-ORCH-026-descriptor", "TR-ACC-IMPORT-ORCH-026-payload", + "TR-ACC-IMPORT-ORCH-027")}, "V2CorePublication": "OUT_OF_SCOPE"} + evidence["versionPin"]["scenarios"]["TR-ACC-IMPORT-ORCH-028"] = { + "result": "NOT_RUN_MISSING_CONTROL", + "reason": "Core admission/dispatch-only pause boundary unavailable"} vector_paths = { "context": args.context_path / "fixtures/trading-normalization/input-identity-v1.json", "core": args.core_path / "tradingreview/application/src/test/resources/trading-normalization/input-identity-v1.json", @@ -522,11 +547,15 @@ def main() -> int: compute_db = None stage_lock = None fault_server = None + pin_proxy = None try: manifest = json.loads((args.fixture_root / "fixture-manifest.json").read_text(encoding="utf-8")) artifacts = {a["role"]: a for a in manifest["artifacts"]} connection = db_connection("public") compute_db = db_connection(os.environ["TRADING_RUNTIME_COMPUTE_SCHEMA"]) + if args.result_version_pin: + from trading_normalization_version_pin import PinProxy + pin_proxy = PinProxy(sys.modules[__name__], args) token = os.environ["TRADING_RUNTIME_ACCESS_TOKEN"] key = os.environ["TRADING_RUNTIME_API_KEY"] csrf = "trading-runtime-csrf" @@ -556,8 +585,9 @@ def main() -> int: current = request(core, "GET", f"{args.core_url}/trading-import-sessions/{session_id}", 200) require(current["canValidate"] and set(current["uploadedArtifactRoles"]) == set(artifacts), "UPLOAD_INCOMPLETE") + validation_key = f"validate-{uuid.uuid4().hex}" validated = request(core, "POST", f"{args.core_url}/trading-import-sessions/{session_id}/validation", - 202, headers={"Idempotency-Key": f"validate-{uuid.uuid4().hex}"}, json={}) + 202, headers={"Idempotency-Key": validation_key}, json={}) require(validated["status"] == "VALIDATING", "SESSION_NOT_VALIDATING") run = poll(lambda: row(connection, """select id::text, input_fingerprint, logical_key, input_identity_json, target_version_set_json, retry_policy_version, @@ -565,6 +595,9 @@ def main() -> int: where import_session_id=%s order by created_at desc limit 1""", (int(session_id),)), lambda r: r["compute_job_id"] is not None or r["status"] == "TERMINAL_FAILURE", args.timeout, "Core dispatch") + evidence["coreRunId"] = run["id"] + if run["compute_job_id"] is not None: + evidence["computeJobId"] = run["compute_job_id"] require(run["compute_job_id"] is not None, str(run["last_failure_code"] or "CORE_DISPATCH_FAILED")) run_id, job_id = run["id"], run["compute_job_id"] @@ -729,8 +762,15 @@ def main() -> int: scenario_started = time.monotonic() verify_publication(args, evidence, core, compute, connection, compute_db, book, session_id, run_id, job_id, run, job, status, - descriptor, payload, result, uploaded, artifacts) + descriptor, payload, result, uploaded, artifacts, pin_proxy) mark_pass(evidence, scenario, scenario_started) + if args.result_version_pin: + scenario = "result_version_pin" + from trading_normalization_version_pin import verify_version_pin + verify_version_pin(sys.modules[__name__], args, evidence, core, compute, + connection, compute_db, pin_proxy, book, account, + session_id, run_id, job_id, dispatch, original, + status, descriptor, payload, artifacts, validation_key) return_code = 0 return return_code @@ -1078,12 +1118,26 @@ def main() -> int: evidence["scenarios"][scenario] = "FAIL" evidence["safeFailureCode"] = error.code if isinstance(error, (ScenarioFailure, IntegrityFailure)) else type(error).__name__ evidence["lastState"] = str(error) if isinstance(error, ScenarioFailure) else type(error).__name__ + if args.result_version_pin: + import traceback + active_pin_scenario = evidence["versionPin"].pop("activeScenario", None) + if active_pin_scenario: + evidence["versionPin"]["scenarios"][active_pin_scenario] = { + "result": "FAIL", "safeFailureCode": evidence["safeFailureCode"]} + evidence["failureLocation"] = [ + {"file": Path(frame.filename).name, "line": frame.lineno, "function": frame.name} + for frame in traceback.extract_tb(error.__traceback__)] if args.publication: evidence["diagnosis"] = publication_diagnosis(connection, compute_db, evidence) + if pin_proxy is not None: + evidence["pinProxy"] = pin_proxy.evidence() + evidence["pinDiagnosis"] = pin_proxy.diagnosis() return_code = 1 else: return_code = 0 finally: + if pin_proxy is not None: + pin_proxy.close() if fault_server is not None: fault_server.shutdown() fault_server.server_close() diff --git a/scripts/trading_normalization_version_pin.py b/scripts/trading_normalization_version_pin.py new file mode 100644 index 0000000..eab5dae --- /dev/null +++ b/scripts/trading_normalization_version_pin.py @@ -0,0 +1,504 @@ +"""Integration-owned pin observations, HTTP faults and pre-pin migration rehearsal. + +The primary producer is always the real Core dispatcher and Compute worker. +Fault responses are scoped to new synthetic sessions; no service code is patched. +Only allowlisted observations leave memory. Signed grants and bodies are never logged. +""" +from __future__ import annotations + +import copy +import hashlib +import json +import os +import subprocess +import threading +import uuid +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import httpx +from psycopg import sql + + +SCHEMA = "resultSchemaVersion" +HASH = "resultHashSpecificationVersion" +V1 = ("TRADING_NORMALIZATION_RESULT_V1", "TRADING_NORMALIZATION_RESULT_HASH_V1") +V2 = ("TRADING_NORMALIZATION_RESULT_V2", "TRADING_NORMALIZATION_RESULT_HASH_V2") + + +def pair(value): + return value.get(SCHEMA), value.get(HASH) + + +def encoded(value): + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode() + + +def fingerprint(value): + """Pre-pin producer fingerprint, also checked against the migrated HTTP replay.""" + value = copy.deepcopy(value) + for grant in value["artifactAccessGrants"]: + grant.pop("accessUri", None) + grant.pop("expiresAt", None) + value["artifactAccessGrants"].sort(key=lambda item: item["role"]) + value["pinnedInstrumentTerms"].sort(key=lambda item: item["instrumentId"]) + return hashlib.sha256(encoded(value)).hexdigest() + + +class PinProxy: + def __init__(self, runtime, args): + self.rt, self.args = runtime, args + self.mode = None + self.faults = {} + self.creates = {} + self.acks = [] + self.errors = [] + self.transport_failures = 0 + self.reject_acks = True + self.injected_ack_failures = 0 + self.core_schema = "public" + proxy = self + + class Handler(BaseHTTPRequestHandler): + def do_GET(self): + self.forward() + + def do_POST(self): + self.forward() + + def forward(self): + try: + if self.headers.get("Transfer-Encoding", "").lower() == "chunked": + chunks = [] + while True: + size = int(self.rfile.readline().strip().split(b";", 1)[0], 16) + if size == 0: + while self.rfile.readline() not in (b"\r\n", b"\n", b""): + pass + break + chunks.append(self.rfile.read(size)) + runtime.require(self.rfile.read(2) == b"\r\n", "PROXY_CHUNK_INVALID") + body = b"".join(chunks) + else: + body = self.rfile.read(int(self.headers.get("Content-Length", "0"))) + headers = {k: v for k, v in self.headers.items() + if k.lower() not in {"host", "connection", "content-length", "accept-encoding", "transfer-encoding"}} + headers["Accept-Encoding"] = "identity" + if self.command == "POST" and self.path.endswith("/acknowledgement"): + ack = json.loads(body) + # A separate connection must see the committed Revision before HTTP ACK. + with runtime.db_connection(proxy.core_schema) as db: + observed = runtime.row(db, """select r.id::text as revision_id,n.id::text as run_id, + n.status as run_status from trading_ledger_revisions r + join trading_normalization_runs n on n.id=r.run_id + join trading_normalization_ack_dispatches a on a.run_id=n.id + where a.payload_id=%s""", (ack["payloadId"],)) + reject_ack = bool(proxy.reject_acks and observed and observed["run_id"] in proxy.creates) + proxy.acks.append({"payloadId": ack["payloadId"], + "disposition": ack.get("disposition"), + "injectedFailure": reject_ack, + "commitVisible": bool(observed and observed["run_status"] == "PUBLISHED"), + "revisionId": observed["revision_id"] if observed else None}) + if reject_ack: + proxy.injected_ack_failures += 1 + self.send_response(503) + self.send_header("Content-Length", "0") + self.end_headers() + return + with httpx.Client(timeout=20, trust_env=False) as upstream: + response = upstream.request(self.command, args.compute_url + self.path, + headers=headers, content=body) + data = response.content + if response.status_code == 202 and self.command == "POST" and self.path.endswith("/normalization-jobs"): + sent, accepted = json.loads(body), response.json() + proxy.creates[sent["coreNormalizationRunId"]] = (sent, accepted) + if proxy.mode and str(sent["coreImportSessionId"]) == proxy.mode[1]: + proxy.faults[accepted["jobId"]] = {"mode": proxy.mode[0], + "sessionId": sent["coreImportSessionId"], + "runId": sent["coreNormalizationRunId"], "payloadId": None, + "descriptorReads": 0, "payloadReads": 0} + proxy.mode = None + if response.status_code == 200 and self.path.endswith("/result"): + job_id = self.path.split("/")[-2] + fault = proxy.faults.get(job_id) + if fault: + descriptor = response.json() + fault["payloadId"] = descriptor["payloadId"] + fault["descriptorReads"] += 1 + if fault["mode"] == "descriptor": + descriptor.update({SCHEMA: V2[0], HASH: V2[1]}) + else: + with httpx.Client(timeout=20, trust_env=False) as upstream: + payload_response = upstream.get(args.compute_url + + "/trading-review/terminal-payloads/" + descriptor["payloadId"], headers=headers) + runtime.require(payload_response.status_code == 200, "FAULT_PAYLOAD_UNAVAILABLE") + payload = payload_response.json() + payload.update({SCHEMA: V2[0], HASH: V2[1]}) + fault["body"] = encoded(payload) + descriptor["byteSize"] = len(fault["body"]) + descriptor["payloadSha256"] = hashlib.sha256(fault["body"]).hexdigest() + data = encoded(descriptor) + for fault in proxy.faults.values(): + if fault["mode"] == "payload" and self.path == "/trading-review/terminal-payloads/" + str(fault["payloadId"]): + data = fault["body"] + fault["payloadReads"] += 1 + self.send_response(response.status_code) + self.send_header("Content-Type", response.headers.get("content-type", "application/json")) + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + except Exception as error: + if isinstance(error, httpx.HTTPError): + # Publication deliberately stops the upstream API to exercise ACK retry. + proxy.transport_failures += 1 + else: + proxy.errors.append(type(error).__name__) + self.send_response(502) + self.send_header("Content-Length", "0") + self.end_headers() + + def log_message(self, *args): + pass + + self.server = ThreadingHTTPServer(("127.0.0.1", args.fixture_port), Handler) + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + self.thread.start() + + def evidence(self): + return {"ackRequests": self.acks, "errors": self.errors, + "upstreamUnavailableRequests": self.transport_failures, + "injectedAckFailures": self.injected_ack_failures, + "creates": {run: {"jobId": accepted["jobId"], "requestPair": pair(sent), + "acceptedPair": pair(accepted)} + for run, (sent, accepted) in self.creates.items()}, + "faults": {job: {k: v for k, v in fault.items() if k != "body"} + for job, fault in self.faults.items()}} + + def diagnosis(self): + values = {} + with self.rt.db_connection("public") as core, self.rt.db_connection(self.args.compute_schema) as compute: + for job, fault in self.faults.items(): + values[job] = self.rt.publication_diagnosis(core, compute, { + "coreRunId": fault["runId"], "coreSessionId": fault["sessionId"], "computeJobId": job}) + return values + + def close(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + + +def verify_version_pin(rt, args, evidence, core, compute, db, compute_db, proxy, book, account, + session_id, run_id, job_id, dispatch, original, status, descriptor, + payload, artifacts, validation_key): + checks = evidence["versionPin"] + + def begin(name): + checks["activeScenario"] = name + checks["scenarios"][name] = {"result": "RUNNING"} + + def passed(name, **values): + checks["scenarios"][name] = {"result": "PASS", **values} + checks.pop("activeScenario", None) + # Preserve completed scenarios even if a later service restart fails. + (args.run_dir / "evidence.json").write_text(json.dumps(evidence, indent=2, default=str) + "\n", encoding="utf-8") + + def stored_job(): + return rt.row(compute_db, "select * from trading_review_jobs where job_id=%s::uuid", (job_id,)) + + begin("V1_PIN_PUBLICATION") + pinned_run = rt.row(db, """select result_schema_version,result_hash_specification_version + from trading_normalization_runs where id=%s::uuid""", (run_id,)) + pinned_dispatch = rt.row(db, """select result_schema_version,result_hash_specification_version + from trading_normalization_dispatches where run_id=%s::uuid""", (run_id,)) + job = stored_job() + sent, accepted = proxy.creates[run_id] + for value in (pinned_run, pinned_dispatch, job): + rt.require((value["result_schema_version"], value["result_hash_specification_version"]) == V1, + "PERSISTED_PIN_DRIFT") + for value in (sent, accepted, original, status, descriptor, payload, payload["resultDigestManifest"]): + rt.require(pair(value) == V1, "HTTP_PIN_DRIFT") + rt.require(len(payload["sectionDigests"]) == 9 and + all(item["hashSpecificationVersion"] == V1[1] for item in payload["sectionDigests"]), + "V1_SECTION_PIN_DRIFT") + rt.require(SCHEMA not in sent["inputIdentity"] and HASH not in sent["inputIdentity"], "RESULT_PIN_IN_INPUT_IDENTITY") + revision = rt.row(db, """select id::text,result_schema_version,result_hash_specification_version + from trading_ledger_revisions where run_id=%s::uuid""", (run_id,)) + rt.require((revision["result_schema_version"], revision["result_hash_specification_version"]) == V1, "REVISION_PIN_DRIFT") + observations = [a for a in proxy.acks if a["payloadId"] == descriptor["payloadId"]] + rt.require(observations and all(a["commitVisible"] and a["revisionId"] == revision["id"] + and a["disposition"] == "PERSISTED" for a in observations), "ACK_BEFORE_COMMIT") + ack_id = rt.row(db, "select id::text from trading_normalization_ack_dispatches where run_id=%s::uuid", (run_id,))["id"] + passed("V1_PIN_PUBLICATION", runId=run_id, jobId=job_id, revisionId=revision["id"], ackId=ack_id, + ackPayloadId=descriptor["payloadId"], pair=V1, ackCommitVisible=True) + + begin("TR-ACC-IMPORT-ORCH-024") + before = stored_job() + grant_before = rt.row(compute_db, """select jsonb_agg(to_jsonb(g) order by grant_revision,artifact_role) as grants + from trading_review_grant_revisions g where job_id=%s::uuid""", (job_id,)) + replay = rt.request(compute, "POST", args.compute_url + "/trading-review/normalization-jobs", 202, + headers={"Idempotency-Key": dispatch["compute_key"]}, json=sent) + rt.require(replay == accepted and pair(replay) == V1, "EXACT_REPLAY_DRIFT") + changed = copy.deepcopy(sent) + changed.update({SCHEMA: V2[0], HASH: V2[1]}) + response = rt.request(compute, "POST", args.compute_url + "/trading-review/normalization-jobs", 409, + headers={"Idempotency-Key": dispatch["compute_key"]}, json=changed) + rt.require(response["code"] == "IDEMPOTENCY_PAYLOAD_CONFLICT" and stored_job() == before, + "PAIR_CONFLICT_CHANGED_JOB") + rt.require(grant_before == rt.row(compute_db, """select jsonb_agg(to_jsonb(g) order by grant_revision,artifact_role) as grants + from trading_review_grant_revisions g where job_id=%s::uuid""", (job_id,)), "PAIR_CONFLICT_CHANGED_GRANTS") + passed("TR-ACC-IMPORT-ORCH-024", jobId=job_id, exactReplay=202, otherPair=409, jobAndGrantsUnchanged=True) + + begin("TR-ACC-IMPORT-ORCH-025") + counts_query = """select (select count(*) from trading_review_jobs) as jobs, + (select count(*) from trading_review_job_idempotency) as idempotency""" + count_before = rt.row(compute_db, counts_query, ()) + cases = [("missing_both", None, None, "REQUEST_INVALID"), + ("missing_schema", None, V1[1], "REQUEST_INVALID"), + ("missing_hash", V1[0], None, "REQUEST_INVALID"), + ("cross_v1_v2", V1[0], V2[1], "RESULT_VERSION_PAIR_INVALID"), + ("cross_v2_v1", V2[0], V1[1], "RESULT_VERSION_PAIR_INVALID"), + ("unsupported_schema", "TRADING_NORMALIZATION_RESULT_FUTURE", V1[1], "RESULT_SCHEMA_UNSUPPORTED"), + ("unsupported_hash", V1[0], "TRADING_NORMALIZATION_RESULT_HASH_FUTURE", "RESULT_SCHEMA_UNSUPPORTED"), + ("unsupported_both", "TRADING_NORMALIZATION_RESULT_FUTURE", "TRADING_NORMALIZATION_RESULT_HASH_FUTURE", "RESULT_SCHEMA_UNSUPPORTED")] + invalid = [] + for name, schema, spec, code in cases: + bad = copy.deepcopy(sent) + for field, value in ((SCHEMA, schema), (HASH, spec)): + if value is None: + bad.pop(field) + else: + bad[field] = value + response = rt.request(compute, "POST", args.compute_url + "/trading-review/normalization-jobs", 422, + headers={"Idempotency-Key": "pin-invalid-" + uuid.uuid4().hex}, json=bad) + rt.require(response["code"] == code, "VERSION_CREATE_WRONG_FAILURE") + rt.require(rt.row(compute_db, counts_query, ()) == count_before, "INVALID_CREATE_PERSISTED") + invalid.append({"case": name, "status": 422, "code": code}) + passed("TR-ACC-IMPORT-ORCH-025", cases=invalid, jobsAndIdempotencyUnchanged=True) + + for mode in ("descriptor", "payload"): + begin("TR-ACC-IMPORT-ORCH-026-" + mode) + fault_session = rt.request(core, "POST", f"{args.core_url}/trading-books/{book['id']}/imports", 201, + headers={"Idempotency-Key": "pin-session-" + uuid.uuid4().hex}, + json={"sourceTimezone": "Asia/Seoul", "reviewTimezone": "UTC", "declaredPositionMode": "ONE_WAY"}) + sid = fault_session["id"] + for role, artifact in artifacts.items(): + rt.upload(core, args.core_url, sid, role, args.fixture_root / artifact["path"], artifact["sha256"]) + proxy.mode = (mode, str(sid)) + rt.request(core, "POST", f"{args.core_url}/trading-import-sessions/{sid}/validation", 202, + headers={"Idempotency-Key": "pin-validate-" + uuid.uuid4().hex}, json={}) + failed = rt.poll(lambda: rt.request(core, "GET", f"{args.core_url}/trading-import-sessions/{sid}", 200), + lambda s: s["status"] in {"PROCESSING_FAILED", "ACCEPTED", "REJECTED"}, args.timeout, "pin fault " + mode) + rt.require(failed["status"] == "PROCESSING_FAILED" and + failed["failure"]["code"] == "RESULT_VERSION_PIN_MISMATCH" and + failed["failure"]["userAction"] == "WAIT_FOR_RECOVERY" and + failed["createdLedgerRevisionId"] is None, "PIN_MISMATCH_NOT_BLOCKED") + fault_job, fault = next((j, f) for j, f in proxy.faults.items() if f["sessionId"] == sid) + state = rt.row(db, """select n.status,d.status as dispatch_status, + (select count(*) from trading_ledger_revisions where run_id=n.id) as revisions, + (select count(*) from trading_normalization_ack_dispatches where run_id=n.id) as acks, + (select latest_ledger_revision_id::text from trading_books where id=n.trading_book_id) as latest + from trading_normalization_runs n join trading_normalization_dispatches d on d.run_id=n.id + where n.id=%s::uuid""", (fault["runId"],)) + rt.require(state == {"status": "TERMINAL_FAILURE", "dispatch_status": "TERMINAL_FAILURE", + "revisions": 0, "acks": 0, "latest": revision["id"]}, "PIN_FAULT_PUBLICATION_DRIFT") + terminal = rt.row(compute_db, """select acknowledged_at from trading_review_terminal_payloads + where job_id=%s::uuid""", (fault_job,)) + rt.require(terminal and terminal["acknowledged_at"] is None and fault["descriptorReads"] > 0 and + (mode != "payload" or fault["payloadReads"] > 0), "PIN_FAULT_NOT_OBSERVED") + passed("TR-ACC-IMPORT-ORCH-026-" + mode, runId=fault["runId"], jobId=fault_job, + payloadId=fault["payloadId"], revisions=0, persistedAcks=0, latestRevisionId=revision["id"]) + passed("TR-ACC-IMPORT-ORCH-026", cases=["descriptor", "payload"]) + + checks["proxy"] = proxy.evidence() + rt.require(not proxy.errors, "PIN_PROXY_ERRORS") + begin("TR-ACC-IMPORT-ORCH-027") + migration_rehearsal(rt, args, checks, core, compute, db, compute_db, proxy, + book, account, session_id, run_id, job_id, sent, dispatch, validation_key) + passed("TR-ACC-IMPORT-ORCH-027", **checks.pop("migration")) + evidence["pinProxy"] = proxy.evidence() + + +def migration_rehearsal(rt, args, checks, core, compute, db, compute_db, proxy, + book, account, session_id, run_id, job_id, sent, dispatch, validation_key): + """Copy ONLY this synthetic publication to pristine pre-pin schemas, then run real SQL. + + No downgrade, DROP COLUMN or trigger bypass occurs in a running service schema. + The services are subsequently restarted against the migrated copies for HTTP replay. + """ + suffix = uuid.uuid4().hex[:12] + core_schema, compute_schema = "tnorm_legacycore" + suffix, "tnorm_legacycompute" + suffix + checks["migration"] = {"coreSchema": core_schema, "computeSchema": compute_schema, + "runId": run_id, "jobId": job_id} + for name in (core_schema, compute_schema): + db.execute(sql.SQL("CREATE SCHEMA {}").format(sql.Identifier(name))) + core_root = args.core_path / "tradingreview/adapter/persistence/src/main/resources/db/migration" + core_migrations = sorted(core_root.glob("V*__*.sql"), key=lambda p: int(p.name.split("__")[0][1:])) + compute_migrations = sorted((args.compute_path / "migrations").glob("*.sql")) + + def migrate(connection, paths): + for path in paths: + if path.name == "0017_pin_trading_normalization_result_pair.sql": + # Match initialize-compute-schema.ps1's existing psql -f execution + # boundary. A single transaction queues the deferred Job/terminal FK + # across UPDATE + ALTER and is not equivalent to that runner. + environment = dict(os.environ, PGPASSWORD=os.environ["TRADING_RUNTIME_DB_PASSWORD"], + PGOPTIONS=f"-c search_path={compute_schema}") + result = subprocess.run([os.environ["TRADING_RUNTIME_PSQL"], + "-h", os.environ["TRADING_RUNTIME_DB_HOST"], "-p", os.environ["TRADING_RUNTIME_DB_PORT"], + "-U", os.environ["TRADING_RUNTIME_DB_USER"], "-d", "refinvest_integration", + "-v", "ON_ERROR_STOP=1", "-v", "VERBOSITY=sqlstate", "-f", str(path)], + env=environment, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, check=False) + migration_log = args.run_dir / "compute-pin-migration.err.log" + migration_log.write_bytes(result.stderr) + checks["migration"]["migrationLogPath"] = str(migration_log) + rt.require(result.returncode == 0, "COMPUTE_PIN_MIGRATION_FAILED") + checks["migration"]["computeMigrationRunner"] = "existing psql -f autocommit boundary" + continue + with connection.transaction(): + connection.execute(path.read_text(encoding="utf-8"), prepare=False) + + def copy_rows(target, source, table, condition, params=(), changes=None): + columns = rt.row(target, """select array_agg(column_name::text order by ordinal_position) as names + from information_schema.columns where table_schema=current_schema() and table_name=%s""", (table,))["names"] + records = source.execute(sql.SQL("SELECT {} FROM {} WHERE ").format( + sql.SQL(",").join(map(sql.Identifier, columns)), sql.Identifier(table)) + sql.SQL(condition), params).fetchall() + for record in records: + values = dict(zip(columns, record)) + if changes: + values.update(changes) + from psycopg.types.json import Jsonb + target.execute(sql.SQL("INSERT INTO {} ({}) VALUES ({})").format(sql.Identifier(table), + sql.SQL(",").join(map(sql.Identifier, columns)), sql.SQL(",").join(sql.Placeholder() for _ in columns)), + [Jsonb(values[c]) if isinstance(values[c], (dict, list)) else values[c] for c in columns]) + + with rt.db_connection(core_schema) as legacy_core, rt.db_connection(compute_schema) as legacy_compute: + migrate(legacy_core, [p for p in core_migrations if int(p.name.split("__")[0][1:]) < 9]) + migrate(legacy_compute, [p for p in compute_migrations if int(p.name[:4]) < 17]) + owner = rt.row(db, "select owner_member_id from trading_books where id=%s", (int(book["id"]),))["owner_member_id"] + copy_rows(legacy_core, db, "members", "id=%s", (owner,)) + legacy_core.execute("CREATE TABLE subscriptions (LIKE public.subscriptions INCLUDING ALL)") + copy_rows(legacy_core, db, "subscriptions", "member_id=%s", (owner,)) + copy_rows(legacy_core, db, "trading_accounts", "id=%s", (int(account["id"]),)) + copy_rows(legacy_core, db, "trading_books", "id=%s", (int(book["id"]),), + {"latest_ledger_revision_id": None}) + copy_rows(legacy_core, db, "trading_import_sessions", "id=%s", (int(session_id),), + {"created_ledger_revision_id": None}) + copy_rows(legacy_core, db, "trading_source_artifacts", "import_session_id=%s", (int(session_id),)) + saved = rt.row(db, "select * from trading_normalization_runs where id=%s::uuid", (run_id,)) + logical_body = {"ownerScope": str(owner), "inputSetHash": saved["input_fingerprint"], + "deletionGeneration": saved["deletion_generation"]} + old_key = hashlib.sha256(encoded(logical_body)).hexdigest() + copy_rows(legacy_core, db, "trading_normalization_runs", "id=%s::uuid", (run_id,), {"logical_key": old_key}) + for table in ("trading_normalization_attempts", "trading_normalization_dispatches", "trading_ledger_revisions", + "trading_normalization_ack_dispatches"): + copy_rows(legacy_core, db, table, "run_id=%s::uuid", (run_id,)) + revision = rt.row(db, "select id::text from trading_ledger_revisions where run_id=%s::uuid", (run_id,))["id"] + legacy_core.execute("UPDATE trading_import_sessions SET created_ledger_revision_id=%s::uuid WHERE id=%s", + (revision, int(session_id))) + legacy_core.execute("UPDATE trading_books SET latest_ledger_revision_id=%s::uuid WHERE id=%s", + (revision, int(book["id"]))) + copy_rows(legacy_core, db, "trading_source_evidence_snapshots", "trading_book_id=%s", (int(book["id"]),)) + copy_rows(legacy_core, db, "trading_records", "trading_book_id=%s", (int(book["id"]),)) + for table in ("trading_ledger_revision_records", "trading_ledger_review_units", "trading_ledger_allocations"): + copy_rows(legacy_core, db, table, "revision_id=%s::uuid", (revision,)) + copy_rows(legacy_core, db, "trading_idempotency", "owner_member_id=%s AND request_key=%s", (owner, validation_key)) + pre_request = copy.deepcopy(sent) + pre_request.pop(SCHEMA) + pre_request.pop(HASH) + # Persistence strips capability locators, as the original pre-pin producer did. + for grant in pre_request["artifactAccessGrants"]: + grant.pop("accessUri", None) + pre_fp = fingerprint(pre_request) + # The production Job -> terminal FK is deferred across the creation transaction. + with legacy_compute.transaction(): + copy_rows(legacy_compute, compute_db, "trading_review_jobs", "job_id=%s::uuid", (job_id,), + {"request_payload": pre_request, "request_fingerprint": pre_fp}) + copy_rows(legacy_compute, compute_db, "trading_review_job_idempotency", "job_id=%s::uuid", (job_id,), + {"request_fingerprint": pre_fp}) + for table in ("trading_review_job_attempts", "trading_review_terminal_payloads", "trading_review_grant_revisions", + "trading_review_job_acknowledgements", "trading_review_attempt_tokens"): + copy_rows(legacy_compute, compute_db, table, "job_id=%s::uuid", (job_id,)) + revision_before = rt.row(legacy_core, "select * from trading_ledger_revisions where run_id=%s::uuid", (run_id,)) + job_before = rt.row(legacy_compute, "select * from trading_review_jobs where job_id=%s::uuid", (job_id,)) + ack_before = rt.row(legacy_core, "select * from trading_normalization_ack_dispatches where run_id=%s::uuid", (run_id,)) + migrate(legacy_core, [p for p in core_migrations if int(p.name.split("__")[0][1:]) == 9]) + migrate(legacy_compute, [p for p in compute_migrations if int(p.name[:4]) == 17]) + for table, conn, field, identifier in ( + ("trading_normalization_runs", legacy_core, "id", run_id), + ("trading_normalization_dispatches", legacy_core, "run_id", run_id), + ("trading_review_jobs", legacy_compute, "job_id", job_id)): + value = conn.execute(sql.SQL("select result_schema_version,result_hash_specification_version from {} where {}=%s::uuid") + .format(sql.Identifier(table), sql.Identifier(field)), (identifier,)).fetchone() + rt.require(value == V1, "MIGRATION_PIN_DRIFT") + logical_body.update({SCHEMA: V1[0], HASH: V1[1]}) + migrated_run = rt.row(legacy_core, "select * from trading_normalization_runs where id=%s::uuid", (run_id,)) + rt.require(migrated_run == saved and + migrated_run["logical_key"] == hashlib.sha256(encoded(logical_body)).hexdigest() and + migrated_run["status"] == saved["status"] and + rt.row(legacy_core, "select compute_key from trading_normalization_dispatches where run_id=%s::uuid", (run_id,))["compute_key"] == dispatch["compute_key"], "MIGRATION_RUN_IDEMPOTENCY_DRIFT") + migrated_job = rt.row(legacy_compute, "select * from trading_review_jobs where job_id=%s::uuid", (job_id,)) + for key, value in job_before.items(): + if key not in {"request_payload", "request_fingerprint"}: + rt.require(migrated_job[key] == value, "MIGRATION_JOB_DRIFT") + rt.require(migrated_job["request_fingerprint"] == fingerprint(sent) and pair(migrated_job["request_payload"]) == V1, + "MIGRATION_FINGERPRINT_DRIFT") + migrated_idempotency = rt.row(legacy_compute, """select request_fingerprint,job_id::text from + trading_review_job_idempotency where job_id=%s::uuid""", (job_id,)) + expected_request = copy.deepcopy(pre_request) + expected_request.update({SCHEMA: V1[0], HASH: V1[1]}) + rt.require(migrated_idempotency == {"request_fingerprint": fingerprint(sent), "job_id": job_id} and + migrated_job["request_payload"] == expected_request, "MIGRATION_REQUEST_OR_IDEMPOTENCY_DRIFT") + rt.require(rt.row(legacy_core, "select * from trading_ledger_revisions where run_id=%s::uuid", (run_id,)) == revision_before and + rt.row(legacy_core, "select * from trading_normalization_ack_dispatches where run_id=%s::uuid", (run_id,)) == ack_before, + "MIGRATION_CHANGED_REVISION_OR_ACK") + + import shutil + log_archive = args.run_dir / "before-migration-logs" + log_archive.mkdir() + for log in (args.test_root / ".integration-data/logs").glob(args.registry + "-*.log"): + shutil.copy2(log, log_archive / log.name) + checks["migration"]["beforeRestartLogDirectory"] = str(log_archive) + scripts = args.test_root / "scripts" + for service in ("core-api", "compute-worker", "compute-api"): + rt.process_script(scripts / "stop-integration.ps1", "-RegistryId", args.registry, "-Service", service) + proxy.core_schema = core_schema + migrated_evidence = args.run_dir / "migration-startup" + migrated_evidence.mkdir() + rt.process_script(scripts / "start-trading-normalization-service.ps1", "-EnvironmentFile", str(args.environment_file), + "-Service", "compute-api", "-ComputePath", str(args.compute_path), "-DatabaseSchema", compute_schema, + "-EvidenceDirectory", str(migrated_evidence), "-RegistryId", args.registry) + rt.process_script(scripts / "start-trading-normalization-service.ps1", "-EnvironmentFile", str(args.environment_file), + "-CorePath", str(args.core_path), "-DatabaseSchema", core_schema, + "-ComputeBaseUrl", f"http://127.0.0.1:{args.fixture_port}", + "-EvidenceDirectory", str(migrated_evidence), "-RegistryId", args.registry) + rt.poll(lambda: rt.health_code(compute, args.compute_url + "/health"), lambda c: c == 200, args.timeout, "migrated Compute startup") + rt.poll(lambda: rt.health_code(core, args.core_url + "/actuator/health"), lambda c: c == 200, 600, "migrated Core startup") + migrated_status = rt.request(compute, "GET", f"{args.compute_url}/trading-review/normalization-jobs/{job_id}", 200) + rt.require(migrated_status["status"] == "COMPLETED" and pair(migrated_status) == V1 and + migrated_status["inputSetHash"] == saved["input_fingerprint"], "MIGRATED_STATUS_DRIFT") + migrated_descriptor = rt.request(compute, "GET", + f"{args.compute_url}/trading-review/normalization-jobs/{job_id}/result", 200) + migrated_payload = compute.get(args.compute_url + "/trading-review/terminal-payloads/" + migrated_descriptor["payloadId"]) + rt.require(migrated_payload.status_code == 200 and pair(migrated_descriptor) == V1 and + pair(migrated_payload.json()) == V1 and + migrated_descriptor["resultHash"] == revision_before["result_hash"] and + hashlib.sha256(migrated_payload.content).hexdigest() == migrated_descriptor["payloadSha256"], + "MIGRATED_TERMINAL_DRIFT") + replay = rt.request(compute, "POST", args.compute_url + "/trading-review/normalization-jobs", 202, + headers={"Idempotency-Key": dispatch["compute_key"]}, json=sent) + rt.require(replay["jobId"] == job_id and pair(replay) == V1, "MIGRATED_COMPUTE_REPLAY_DRIFT") + rt.request(core, "POST", f"{args.core_url}/trading-import-sessions/{session_id}/validation", 202, + headers={"Idempotency-Key": validation_key}, json={}) + public = rt.request(core, "GET", f"{args.core_url}/trading-import-sessions/{session_id}", 200) + rt.require(public["status"] == "ACCEPTED" and public["createdLedgerRevisionId"] == revision, "MIGRATED_CORE_REPLAY_DRIFT") + invalid = rt.request(compute, "POST", args.compute_url + "/trading-review/normalization-jobs", 422, + headers={"Idempotency-Key": "post-migration-" + uuid.uuid4().hex}, json=pre_request) + rt.require(invalid["code"] == "REQUEST_INVALID", "MIGRATION_ENABLED_IMPLICIT_DEFAULT") + rt.require(rt.row(legacy_core, "select * from trading_ledger_revisions where run_id=%s::uuid", (run_id,)) == revision_before and + rt.row(legacy_core, "select * from trading_normalization_ack_dispatches where run_id=%s::uuid", (run_id,)) == ack_before, + "RESTART_CHANGED_REVISION_OR_ACK") + rt.require(rt.row(legacy_core, "select count(*) as total from trading_normalization_runs", ())["total"] == 1 and + rt.row(legacy_compute, "select count(*) as total from trading_review_jobs", ())["total"] == 1, + "MIGRATION_REPLAY_DUPLICATED") + checks["migration"].update(revisionId=revision, coreMigration="V9", computeMigration="0017", + serviceRestart=True, explicitReplay=True, revisionAndAckUnchanged=True)