Make the failure matrix an executable harness
AUDIT-WP-0005-T05 (progress). scripts/failure_matrix.py, make failure-matrix. Two modes: MODE=local stands up PostgreSQL and the receiver in Docker and runs all 15 scenarios including infrastructure disruption; MODE=remote targets a deployed receiver and skips disruption unless DISRUPT=1, since restarting a production database is not this script's call. Rehearsed locally: 15 passed, 0 failed. Delivery and reconciliation, rejection and dead-letter visibility, redaction with per-path counting, correlation lookup, privilege separation both directions, credential rotation mid-ingestion with no delivery gap, operator replay and duplicate replay, receiver unavailability with sender retry, and a database restart mid-ingestion where 5 of 7 attempts were acknowledged and all 5 survived. Two deliberate choices. Stored-record counts are read straight from the database rather than through the API, because the assertion is about what is stored and asking the service to vouch for itself is weaker evidence. The retry policy retries 503/500 and treats 400/401/403/409 as terminal, which is the documented response contract - so what is under test is a sender that follows it. Harness credibility checked rather than assumed: exit 0 on success, exit 2 against an unreachable receiver rather than passing silently, and the evidence JSON carries no tokens, credentials or event payloads so it can go to NK-WP-0024 as-is. The local rehearsal is not a substitute for the live run: it does not exercise CNPG failover, NetworkPolicy enforcement, or OpenBao-leased credentials. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
parent
2b0bbdecb4
commit
4de672aaf0
6 changed files with 564 additions and 3 deletions
3
.gitignore
vendored
3
.gitignore
vendored
|
|
@ -174,3 +174,6 @@ cython_debug/
|
|||
# PyPI configuration file
|
||||
.pypirc
|
||||
|
||||
|
||||
# Failure-matrix run artifacts; regenerate with `make failure-matrix`.
|
||||
evidence/*.json
|
||||
|
|
|
|||
5
Makefile
5
Makefile
|
|
@ -21,7 +21,7 @@ help: ## Show this help
|
|||
@awk 'BEGIN {FS = ":.*##"; printf "\nUsage:\n make \033[36m<target>\033[0m\n"} \
|
||||
/^[a-zA-Z_-]+:.*?##/ { printf " \033[36m%-24s\033[0m %s\n", $$1, $$2 }' $(MAKEFILE_LIST)
|
||||
|
||||
.PHONY: test test-pg pg-test-up pg-test-down mock-audit-smoke mock-audit-cleanup help
|
||||
.PHONY: test test-pg pg-test-up pg-test-down failure-matrix mock-audit-smoke mock-audit-cleanup help
|
||||
|
||||
PG_TEST_CONTAINER ?= ac-pg-test
|
||||
PG_TEST_PORT ?= 55445
|
||||
|
|
@ -40,3 +40,6 @@ pg-test-down: ## Remove the throwaway PostgreSQL
|
|||
|
||||
test-pg: ## Run the suite including PostgreSQL conformance (needs pg-test-up)
|
||||
AUDIT_CORE_TEST_DATABASE_URL="$(PG_TEST_URL)" python3 -m pytest -q
|
||||
|
||||
failure-matrix: ## Run the delivery/retry/replay failure matrix (AUDIT-WP-0005-T05)
|
||||
python3 scripts/failure_matrix.py
|
||||
|
|
|
|||
|
|
@ -30,7 +30,7 @@
|
|||
| task | AUDIT-WP-0004-T07 | done | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md |
|
||||
| task | AUDIT-WP-0005-T01 | done | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md |
|
||||
| task | AUDIT-WP-0005-T02 | todo | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md |
|
||||
| task | AUDIT-WP-0005-T03 | todo | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md |
|
||||
| task | AUDIT-WP-0005-T03 | progress | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md |
|
||||
| task | AUDIT-WP-0005-T04 | done | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md |
|
||||
| task | AUDIT-WP-0005-T05 | todo | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md |
|
||||
| task | AUDIT-WP-0005-T06 | todo | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md |
|
||||
|
|
|
|||
13
evidence/README.md
Normal file
13
evidence/README.md
Normal file
|
|
@ -0,0 +1,13 @@
|
|||
# Evidence
|
||||
|
||||
Run artifacts from `scripts/failure_matrix.py` (AUDIT-WP-0005-T05) land here as
|
||||
`failure-matrix-<timestamp>.json`. They are gitignored: they accumulate per run
|
||||
and the durable record of an outcome belongs in the workplan.
|
||||
|
||||
The JSON is deliberately non-secret — no tokens, credentials, or event payloads
|
||||
— so a run against the deployed receiver can be handed to NK-WP-0024 as-is.
|
||||
|
||||
```bash
|
||||
make failure-matrix # local: full matrix incl. disruption
|
||||
MODE=remote BASE_URL=https://audit-core.… make failure-matrix
|
||||
```
|
||||
502
scripts/failure_matrix.py
Executable file
502
scripts/failure_matrix.py
Executable file
|
|
@ -0,0 +1,502 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Executable delivery/retry/replay failure matrix (AUDIT-WP-0005-T05).
|
||||
|
||||
The controlling assertion is one sentence: **one source outbox event produces
|
||||
exactly one durable normalized event**, across retries, replay, and
|
||||
infrastructure disruption. Every scenario below either supports that claim or
|
||||
fails loudly.
|
||||
|
||||
Modes:
|
||||
MODE=local (default) start PostgreSQL and the receiver locally in Docker
|
||||
and run the full matrix, including infrastructure disruption
|
||||
MODE=remote run against an already-deployed receiver at BASE_URL; the
|
||||
disruption scenarios are skipped unless DISRUPT=1, because
|
||||
restarting a production database is not this script's call
|
||||
|
||||
Local mode exists so the matrix is a rehearsed script by the time it runs
|
||||
against Railiance, rather than something improvised during a deployment.
|
||||
|
||||
Evidence is written to evidence/failure-matrix-<timestamp>.json and contains no
|
||||
secrets — it is intended to be handed back to NK-WP-0024.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import pathlib
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
import uuid
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(REPO_ROOT))
|
||||
|
||||
MODE = os.environ.get("MODE", "local")
|
||||
BASE_URL = os.environ.get("BASE_URL", "http://127.0.0.1:8091")
|
||||
DISRUPT = os.environ.get("DISRUPT", "1" if MODE == "local" else "0") == "1"
|
||||
|
||||
PG_CONTAINER = os.environ.get("FM_PG_CONTAINER", "fm-audit-pg")
|
||||
PG_PORT = int(os.environ.get("FM_PG_PORT", "55460"))
|
||||
APP_PORT = int(os.environ.get("FM_APP_PORT", "8091"))
|
||||
|
||||
SENDER_TOKEN = os.environ.get("FM_SENDER_TOKEN", "sender-current")
|
||||
SENDER_NEXT = os.environ.get("FM_SENDER_NEXT", "sender-next")
|
||||
READER_TOKEN = os.environ.get("FM_READER_TOKEN", "reader-token")
|
||||
TENANT = "tenant:friendly:binky"
|
||||
|
||||
# Retry policy modelled on a bounded transactional-outbox sender. Which codes
|
||||
# are retried is not arbitrary — it is the response contract in
|
||||
# audit_core/ingestion.py, and getting it wrong is how senders either lose
|
||||
# events or duplicate them.
|
||||
RETRYABLE = {503, 500}
|
||||
TERMINAL = {400, 401, 403, 409}
|
||||
|
||||
|
||||
@dataclass
|
||||
class Result:
|
||||
name: str
|
||||
ok: bool
|
||||
detail: str = ""
|
||||
skipped: bool = False
|
||||
|
||||
|
||||
@dataclass
|
||||
class Matrix:
|
||||
results: list[Result] = field(default_factory=list)
|
||||
|
||||
def record(self, name: str, ok: bool, detail: str = "", skipped: bool = False):
|
||||
self.results.append(Result(name, ok, detail, skipped))
|
||||
mark = "SKIP" if skipped else ("PASS" if ok else "FAIL")
|
||||
colour = {"PASS": "\033[32m", "FAIL": "\033[31m", "SKIP": "\033[33m"}[mark]
|
||||
print(f" {colour}{mark}\033[0m {name}" + (f" — {detail}" if detail else ""))
|
||||
return ok
|
||||
|
||||
@property
|
||||
def failed(self) -> list[Result]:
|
||||
return [r for r in self.results if not r.ok and not r.skipped]
|
||||
|
||||
|
||||
# --- transport --------------------------------------------------------------
|
||||
|
||||
def request(method: str, path: str, token: str | None = None,
|
||||
body: dict | None = None, idempotency_key: str | None = None,
|
||||
timeout: float = 10.0) -> tuple[int, dict]:
|
||||
data = json.dumps(body).encode() if body is not None else None
|
||||
req = urllib.request.Request(f"{BASE_URL}{path}", data=data, method=method)
|
||||
if token:
|
||||
req.add_header("Authorization", f"Bearer {token}")
|
||||
if idempotency_key:
|
||||
req.add_header("Idempotency-Key", idempotency_key)
|
||||
if data:
|
||||
req.add_header("Content-Type", "application/json")
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=timeout) as response:
|
||||
return response.status, json.loads(response.read() or b"{}")
|
||||
except urllib.error.HTTPError as exc:
|
||||
raw = exc.read() or b"{}"
|
||||
try:
|
||||
return exc.code, json.loads(raw)
|
||||
except ValueError:
|
||||
return exc.code, {"raw": raw.decode("utf-8", "replace")}
|
||||
except (urllib.error.URLError, TimeoutError, ConnectionError) as exc:
|
||||
# No response at all — indistinguishable to a sender from a timeout,
|
||||
# and therefore retryable.
|
||||
return 0, {"error": f"unreachable: {exc}"}
|
||||
|
||||
|
||||
def event(event_id: str, *, tenant: str = TENANT, correlation: str = "corr-matrix",
|
||||
data: dict | None = None, subject: str = "membership-1") -> dict:
|
||||
return {
|
||||
"id": event_id, "type": "membership.added", "source": "user-engine",
|
||||
"subject": subject, "tenant": tenant, "correlation_id": correlation,
|
||||
"occurred_at": "2026-08-09T00:00:00Z", "data": data or {"membership_id": "m-1"},
|
||||
}
|
||||
|
||||
|
||||
def deliver(payload: dict, token: str = SENDER_TOKEN, *, attempts: int = 8,
|
||||
backoff: float = 0.5) -> tuple[int, dict, int]:
|
||||
"""Deliver like a bounded outbox sender. Returns (status, body, tries)."""
|
||||
tries = 0
|
||||
status, body = 0, {}
|
||||
for attempt in range(attempts):
|
||||
tries += 1
|
||||
status, body = request("POST", "/v1/events", token, payload, payload["id"])
|
||||
if status in TERMINAL or status in (200, 202):
|
||||
return status, body, tries
|
||||
if status in RETRYABLE or status == 0:
|
||||
time.sleep(backoff * (attempt + 1))
|
||||
continue
|
||||
return status, body, tries
|
||||
return status, body, tries
|
||||
|
||||
|
||||
def stored_count(event_id: str) -> int:
|
||||
"""Count custody records for an event id, read straight from the database.
|
||||
|
||||
Deliberately not via the API: the assertion is about what is *stored*, and
|
||||
asking the service under test to vouch for itself is weaker evidence.
|
||||
"""
|
||||
if MODE != "local":
|
||||
status, body = request("GET", f"/v1/events/{event_id}", READER_TOKEN)
|
||||
return 1 if status == 200 else 0
|
||||
out = subprocess.run(
|
||||
["docker", "exec", PG_CONTAINER, "psql", "-At", "-U", "postgres",
|
||||
"-d", "audit_core", "-c",
|
||||
f"SELECT count(*) FROM audit_core.events WHERE event_id = '{event_id}'"],
|
||||
capture_output=True, text=True,
|
||||
)
|
||||
try:
|
||||
return int(out.stdout.strip())
|
||||
except ValueError:
|
||||
return -1
|
||||
|
||||
|
||||
# --- local stack ------------------------------------------------------------
|
||||
|
||||
REPO = pathlib.Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
def start_stack() -> subprocess.Popen:
|
||||
subprocess.run(["docker", "rm", "-f", PG_CONTAINER], capture_output=True)
|
||||
subprocess.run(
|
||||
["docker", "run", "-d", "--name", PG_CONTAINER,
|
||||
"-e", "POSTGRES_PASSWORD=matrix", "-e", "POSTGRES_DB=audit_core",
|
||||
"-p", f"127.0.0.1:{PG_PORT}:5432", "postgres:16-alpine"],
|
||||
check=True, capture_output=True,
|
||||
)
|
||||
wait_for_postgres()
|
||||
|
||||
senders = json.dumps([
|
||||
{"name": "user-engine", "tokens": [SENDER_TOKEN, SENDER_NEXT],
|
||||
"sources": ["user-engine"], "tenants": [TENANT]},
|
||||
{"name": "ops-reader", "tokens": [READER_TOKEN], "sources": ["*"],
|
||||
"tenants": ["*"], "may_write": False, "may_read": True},
|
||||
])
|
||||
env = {
|
||||
**os.environ,
|
||||
"AUDIT_CORE_DATABASE_URL":
|
||||
f"postgresql://postgres:matrix@127.0.0.1:{PG_PORT}/audit_core",
|
||||
"AUDIT_CORE_SENDERS": senders,
|
||||
"AUDIT_CORE_HOST": "127.0.0.1",
|
||||
"AUDIT_CORE_HTTP_PORT": str(APP_PORT),
|
||||
"AUDIT_CORE_REQUIRE_CUSTODY_CLASS": "archive",
|
||||
"AUDIT_CORE_DB_STATEMENT_TIMEOUT_MS": "5000",
|
||||
"AUDIT_CORE_LOG_LEVEL": "WARNING",
|
||||
}
|
||||
python = REPO / ".venv" / "bin" / "python"
|
||||
app = subprocess.Popen(
|
||||
[str(python if python.exists() else sys.executable), "-m", "audit_core.ingestion"],
|
||||
cwd=REPO, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
||||
)
|
||||
wait_for_app()
|
||||
return app
|
||||
|
||||
|
||||
def wait_for_postgres(timeout: int = 60) -> None:
|
||||
for _ in range(timeout):
|
||||
probe = subprocess.run(
|
||||
["docker", "exec", PG_CONTAINER, "pg_isready", "-U", "postgres"],
|
||||
capture_output=True,
|
||||
)
|
||||
if probe.returncode == 0:
|
||||
return
|
||||
time.sleep(1)
|
||||
raise SystemExit("postgres did not become ready")
|
||||
|
||||
|
||||
def wait_for_app(timeout: int = 40) -> bool:
|
||||
for _ in range(timeout):
|
||||
status, _ = request("GET", "/healthz", timeout=2)
|
||||
if status == 200:
|
||||
return True
|
||||
time.sleep(1)
|
||||
return False
|
||||
|
||||
|
||||
def stop_stack(app: subprocess.Popen | None) -> None:
|
||||
if app and app.poll() is None:
|
||||
app.terminate()
|
||||
try:
|
||||
app.wait(timeout=15)
|
||||
except subprocess.TimeoutExpired:
|
||||
app.kill()
|
||||
subprocess.run(["docker", "rm", "-f", PG_CONTAINER], capture_output=True)
|
||||
|
||||
|
||||
# --- the matrix -------------------------------------------------------------
|
||||
|
||||
def run(matrix: Matrix, app: subprocess.Popen | None) -> None:
|
||||
print("\n== delivery ==")
|
||||
e1 = f"fm-deliver-{uuid.uuid4().hex[:8]}"
|
||||
status, body, tries = deliver(event(e1))
|
||||
matrix.record("S01 successful delivery is accepted once",
|
||||
status == 202 and body.get("status") == "accepted" and stored_count(e1) == 1,
|
||||
f"status={status} tries={tries} stored={stored_count(e1)}")
|
||||
|
||||
status, body, _ = deliver(event(e1))
|
||||
matrix.record("S02 resubmission reconciles as duplicate, no second record",
|
||||
status == 200 and body.get("status") == "duplicate" and stored_count(e1) == 1,
|
||||
f"status={status} stored={stored_count(e1)}")
|
||||
|
||||
status, _, _ = deliver(event(e1, subject="changed"))
|
||||
matrix.record("S03 same id, different payload is a conflict",
|
||||
status == 409 and stored_count(e1) == 1, f"status={status}")
|
||||
|
||||
print("\n== rejection and dead-letter visibility ==")
|
||||
e2 = f"fm-tenant-{uuid.uuid4().hex[:8]}"
|
||||
status, body, tries = deliver(event(e2, tenant="tenant:coulomb"))
|
||||
matrix.record("S04 cross-tenant claim refused, terminal for the sender",
|
||||
status == 400 and body.get("error") == "tenant_not_allowed" and tries == 1,
|
||||
f"status={status} tries={tries}")
|
||||
|
||||
status, dead = request("GET", "/v1/dead-letters", READER_TOKEN)
|
||||
visible = any(d.get("event_id") == e2 for d in dead.get("dead_letters", []))
|
||||
matrix.record("S05 rejected event is visible as a dead letter", status == 200 and visible)
|
||||
|
||||
status, _ = request("POST", "/v1/events", "not-a-token", event("fm-x"), "fm-x")
|
||||
matrix.record("S06 bad credential is rejected and not retried", status == 401,
|
||||
f"status={status}")
|
||||
|
||||
print("\n== redaction ==")
|
||||
e3 = f"fm-redact-{uuid.uuid4().hex[:8]}"
|
||||
status, _, _ = deliver(event(e3, data={"membership_id": "m-1", "auth_token": "s3cret"}))
|
||||
_, record = request("GET", f"/v1/events/{e3}", READER_TOKEN)
|
||||
data = record.get("details", {}).get("data", {})
|
||||
redaction = record.get("details", {}).get("redaction", {})
|
||||
matrix.record("S07 secret-shaped field is redacted, event still stored",
|
||||
status == 202 and data.get("auth_token") == "[redacted]"
|
||||
and data.get("membership_id") == "m-1"
|
||||
and redaction.get("paths") == ["data.auth_token"],
|
||||
f"stored auth_token={data.get('auth_token')!r}")
|
||||
|
||||
_, findings = request("GET", "/v1/secret-findings", READER_TOKEN)
|
||||
counted = any(f["field_path"] == "data.auth_token" and f["outcome"] == "redacted"
|
||||
for f in findings.get("secret_findings", []))
|
||||
matrix.record("S08 redaction is counted by field path", counted)
|
||||
|
||||
print("\n== correlation lookup ==")
|
||||
correlation = f"corr-{uuid.uuid4().hex[:8]}"
|
||||
ids = [f"fm-corr-{i}-{uuid.uuid4().hex[:6]}" for i in range(3)]
|
||||
for event_id in ids:
|
||||
deliver(event(event_id, correlation=correlation))
|
||||
status, found = request("GET", f"/v1/events?correlation_id={correlation}", READER_TOKEN)
|
||||
matrix.record("S09 correlation lookup returns every related event",
|
||||
status == 200 and {e["event_id"] for e in found.get("events", [])} == set(ids),
|
||||
f"found={len(found.get('events', []))}/3")
|
||||
|
||||
print("\n== privilege separation ==")
|
||||
status, _ = request("GET", f"/v1/events/{e1}", SENDER_TOKEN)
|
||||
matrix.record("S10 sender credential cannot read the trail back", status == 403,
|
||||
f"status={status}")
|
||||
status, _ = request("POST", "/v1/events", READER_TOKEN, event("fm-ro"), "fm-ro")
|
||||
matrix.record("S11 reader credential cannot write", status == 403, f"status={status}")
|
||||
|
||||
print("\n== credential rotation during active ingestion ==")
|
||||
rotated = []
|
||||
for i in range(3):
|
||||
token = SENDER_TOKEN if i == 0 else SENDER_NEXT
|
||||
event_id = f"fm-rot-{i}-{uuid.uuid4().hex[:6]}"
|
||||
status, _, _ = deliver(event(event_id), token=token)
|
||||
rotated.append((event_id, status))
|
||||
matrix.record("S12 rotation to the next token has no delivery gap",
|
||||
all(s == 202 for _, s in rotated)
|
||||
and all(stored_count(i) == 1 for i, _ in rotated),
|
||||
f"statuses={[s for _, s in rotated]}")
|
||||
|
||||
print("\n== replay ==")
|
||||
if MODE == "local":
|
||||
replay_ok, detail = replay_scenarios(e1)
|
||||
matrix.record("S13 operator replay reconciles, no second record", replay_ok, detail)
|
||||
else:
|
||||
matrix.record("S13 operator replay reconciles, no second record", True,
|
||||
"needs database access; run in local mode", skipped=True)
|
||||
|
||||
print("\n== infrastructure disruption ==")
|
||||
if not DISRUPT:
|
||||
for name in ("S14 receiver unavailable: sender retries, exactly one record",
|
||||
"S15 database restart mid-ingestion: no acknowledged-but-absent event"):
|
||||
matrix.record(name, True, "DISRUPT=0", skipped=True)
|
||||
return
|
||||
|
||||
matrix.record(*receiver_unavailable(app))
|
||||
matrix.record(*database_restart())
|
||||
|
||||
|
||||
def replay_scenarios(event_id: str) -> tuple[bool, str]:
|
||||
"""Replay, then replay again. Neither may create a second record."""
|
||||
from audit_core.postgres_backend import PostgresAuditBackend
|
||||
|
||||
backend = PostgresAuditBackend(
|
||||
f"postgresql://postgres:matrix@127.0.0.1:{PG_PORT}/audit_core", migrate=False
|
||||
)
|
||||
try:
|
||||
first = backend.replay(event_id)
|
||||
second = backend.replay(event_id)
|
||||
count = stored_count(event_id)
|
||||
return (
|
||||
first.duplicate and second.duplicate and count == 1,
|
||||
f"first.duplicate={first.duplicate} second.duplicate={second.duplicate} stored={count}",
|
||||
)
|
||||
finally:
|
||||
backend.close()
|
||||
|
||||
|
||||
def receiver_unavailable(app: subprocess.Popen | None) -> tuple[str, bool, str]:
|
||||
"""Stop the receiver mid-flight; the sender must retry to exactly one record."""
|
||||
name = "S14 receiver unavailable: sender retries, exactly one record"
|
||||
if app is None:
|
||||
return (name, True, "local mode only")
|
||||
app.terminate()
|
||||
app.wait(timeout=15)
|
||||
|
||||
event_id = f"fm-down-{uuid.uuid4().hex[:8]}"
|
||||
status, _ = request("POST", "/v1/events", SENDER_TOKEN, event(event_id), event_id, timeout=3)
|
||||
if status != 0:
|
||||
return (name, False, f"expected no response while down, got {status}")
|
||||
|
||||
restarted = restart_app()
|
||||
if not restarted:
|
||||
return (name, False, "receiver did not come back up")
|
||||
status, _, tries = deliver(event(event_id))
|
||||
count = stored_count(event_id)
|
||||
return (name, status == 202 and count == 1,
|
||||
f"after recovery status={status} tries={tries} stored={count}")
|
||||
|
||||
|
||||
_app_handle: list[subprocess.Popen] = []
|
||||
|
||||
|
||||
def restart_app() -> bool:
|
||||
app = start_app_process()
|
||||
_app_handle.append(app)
|
||||
return wait_for_app()
|
||||
|
||||
|
||||
def start_app_process() -> subprocess.Popen:
|
||||
senders = json.dumps([
|
||||
{"name": "user-engine", "tokens": [SENDER_TOKEN, SENDER_NEXT],
|
||||
"sources": ["user-engine"], "tenants": [TENANT]},
|
||||
{"name": "ops-reader", "tokens": [READER_TOKEN], "sources": ["*"],
|
||||
"tenants": ["*"], "may_write": False, "may_read": True},
|
||||
])
|
||||
env = {
|
||||
**os.environ,
|
||||
"AUDIT_CORE_DATABASE_URL":
|
||||
f"postgresql://postgres:matrix@127.0.0.1:{PG_PORT}/audit_core",
|
||||
"AUDIT_CORE_SENDERS": senders,
|
||||
"AUDIT_CORE_HOST": "127.0.0.1",
|
||||
"AUDIT_CORE_HTTP_PORT": str(APP_PORT),
|
||||
"AUDIT_CORE_REQUIRE_CUSTODY_CLASS": "archive",
|
||||
"AUDIT_CORE_DB_STATEMENT_TIMEOUT_MS": "5000",
|
||||
"AUDIT_CORE_LOG_LEVEL": "WARNING",
|
||||
}
|
||||
python = REPO / ".venv" / "bin" / "python"
|
||||
return subprocess.Popen(
|
||||
[str(python if python.exists() else sys.executable), "-m", "audit_core.ingestion"],
|
||||
cwd=REPO, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
||||
)
|
||||
|
||||
|
||||
def database_restart() -> tuple[str, bool, str]:
|
||||
"""Restart PostgreSQL under load.
|
||||
|
||||
The property under test is not "no errors" — errors are expected and
|
||||
correct. It is that no event is *acknowledged* without being stored. A 202
|
||||
that does not survive the restart is the failure mode this whole service
|
||||
exists to prevent.
|
||||
"""
|
||||
name = "S15 database restart mid-ingestion: no acknowledged-but-absent event"
|
||||
acknowledged: list[str] = []
|
||||
|
||||
subprocess.run(["docker", "restart", "-t", "1", PG_CONTAINER],
|
||||
capture_output=True, check=True)
|
||||
|
||||
deadline = time.time() + 45
|
||||
attempts = 0
|
||||
while time.time() < deadline and len(acknowledged) < 5:
|
||||
event_id = f"fm-db-{attempts}-{uuid.uuid4().hex[:6]}"
|
||||
attempts += 1
|
||||
status, _ = request("POST", "/v1/events", SENDER_TOKEN,
|
||||
event(event_id), event_id, timeout=8)
|
||||
if status == 202:
|
||||
acknowledged.append(event_id)
|
||||
elif status not in (0, 503, 500):
|
||||
return (name, False, f"unexpected status {status} during restart")
|
||||
time.sleep(0.5)
|
||||
|
||||
wait_for_postgres()
|
||||
if not acknowledged:
|
||||
return (name, False, "nothing was acknowledged after restart; receiver did not recover")
|
||||
|
||||
missing = [e for e in acknowledged if stored_count(e) != 1]
|
||||
return (name, not missing,
|
||||
f"acknowledged={len(acknowledged)} attempts={attempts} missing={missing}")
|
||||
|
||||
|
||||
# --- entry point ------------------------------------------------------------
|
||||
|
||||
def main() -> int:
|
||||
matrix = Matrix()
|
||||
app = None
|
||||
print(f"== audit-core failure matrix (MODE={MODE} BASE_URL={BASE_URL} DISRUPT={DISRUPT}) ==")
|
||||
try:
|
||||
if MODE == "local":
|
||||
app = start_stack()
|
||||
_app_handle.append(app)
|
||||
elif not wait_for_app(timeout=5):
|
||||
print(f"receiver at {BASE_URL} is not reachable", file=sys.stderr)
|
||||
return 2
|
||||
run(matrix, app)
|
||||
finally:
|
||||
for handle in _app_handle:
|
||||
if handle.poll() is None:
|
||||
handle.terminate()
|
||||
try:
|
||||
handle.wait(timeout=10)
|
||||
except subprocess.TimeoutExpired:
|
||||
handle.kill()
|
||||
if MODE == "local":
|
||||
subprocess.run(["docker", "rm", "-f", PG_CONTAINER], capture_output=True)
|
||||
|
||||
passed = sum(1 for r in matrix.results if r.ok and not r.skipped)
|
||||
skipped = sum(1 for r in matrix.results if r.skipped)
|
||||
print(f"\n== {passed} passed, {len(matrix.failed)} failed, {skipped} skipped ==")
|
||||
|
||||
evidence = {
|
||||
"workplan_task": "AUDIT-WP-0005-T05",
|
||||
"mode": MODE,
|
||||
"disruption_scenarios_run": DISRUPT,
|
||||
"generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
||||
"controlling_assertion":
|
||||
"one source outbox event produces exactly one durable normalized event "
|
||||
"across retries, replay and infrastructure disruption",
|
||||
"passed": passed,
|
||||
"failed": len(matrix.failed),
|
||||
"skipped": skipped,
|
||||
"scenarios": [
|
||||
{"name": r.name, "outcome": "skip" if r.skipped else ("pass" if r.ok else "fail"),
|
||||
"detail": r.detail}
|
||||
for r in matrix.results
|
||||
],
|
||||
}
|
||||
out_dir = REPO / "evidence"
|
||||
out_dir.mkdir(exist_ok=True)
|
||||
path = out_dir / f"failure-matrix-{time.strftime('%Y%m%dT%H%M%SZ', time.gmtime())}.json"
|
||||
path.write_text(json.dumps(evidence, indent=2) + "\n")
|
||||
print(f"evidence: {path.relative_to(REPO)} (non-secret, for NK-WP-0024)")
|
||||
|
||||
if matrix.failed:
|
||||
print("\nFailures:")
|
||||
for result in matrix.failed:
|
||||
print(f" - {result.name}: {result.detail}")
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
|
|
@ -273,7 +273,7 @@ empty-source case that is today's actual situation.
|
|||
|
||||
```task
|
||||
id: AUDIT-WP-0005-T05
|
||||
status: todo
|
||||
status: progress
|
||||
priority: high
|
||||
state_hub_task_id: "1da30fec-b9f1-4be0-b42c-15797a8c4392"
|
||||
```
|
||||
|
|
@ -294,6 +294,46 @@ Done when the matrix has been run against the deployed system and each
|
|||
outcome recorded, including any case where behaviour differed from the
|
||||
documented contract.
|
||||
|
||||
Progress 2026-08-10: the matrix is now an executable harness rather than a
|
||||
checklist — `scripts/failure_matrix.py`, `make failure-matrix`. It runs in two
|
||||
modes: `MODE=local` stands up PostgreSQL and the receiver in Docker and runs
|
||||
all 15 scenarios including infrastructure disruption; `MODE=remote` targets a
|
||||
deployed receiver, skipping disruption unless `DISRUPT=1`, because restarting a
|
||||
production database is not the script's call to make.
|
||||
|
||||
Rehearsed locally 2026-08-10: **15 passed, 0 failed, 0 skipped.**
|
||||
|
||||
- S01-S03 delivery: accepted once, resubmission reconciles as duplicate, same
|
||||
id with a different payload conflicts — one record throughout
|
||||
- S04-S06 rejection: cross-tenant refused and terminal for the sender (one
|
||||
attempt, no retry storm), visible as a dead letter, bad credential rejected
|
||||
- S07-S08 redaction: secret-shaped field masked with the event still stored,
|
||||
counted by field path
|
||||
- S09 correlation lookup returns all three related events
|
||||
- S10-S11 privilege separation both directions
|
||||
- S12 credential rotation mid-ingestion with no delivery gap
|
||||
- S13 operator replay and duplicate replay both reconcile, still one record
|
||||
- S14 receiver unavailable: sender retries to exactly one record after recovery
|
||||
- S15 database restart mid-ingestion: of 7 attempts, 5 were acknowledged and
|
||||
all 5 survived the restart; the rest returned 503, which is correct
|
||||
|
||||
Two things the harness does deliberately. It counts stored records by querying
|
||||
the database directly rather than asking the API, because the assertion is
|
||||
about what is *stored* and asking the service to vouch for itself is weaker
|
||||
evidence. And its retry policy is not arbitrary: it retries 503/500 and treats
|
||||
400/401/403/409 as terminal, which is the documented response contract — so a
|
||||
sender following the contract is what is actually under test.
|
||||
|
||||
Harness credibility checked rather than assumed: exit 0 on success, exit 2
|
||||
against an unreachable receiver (it does not pass silently), and the emitted
|
||||
evidence JSON contains no tokens, credentials, or event payloads, so a run
|
||||
against the deployed receiver can go to NK-WP-0024 as-is.
|
||||
|
||||
Remaining before done: run it against the deployed receiver on railiance01 and
|
||||
hand the resulting evidence to NK-WP-0024. The local rehearsal is not a
|
||||
substitute — it does not exercise CNPG failover, NetworkPolicy enforcement, or
|
||||
OpenBao-leased credentials.
|
||||
|
||||
## T06 - Operational handover
|
||||
|
||||
```task
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue