"""G6 -- one differential test per pathology (phase-0a-test-suite-spec.md §6). Every test: set chaos-clean, warm up, measure baseline, enable exactly one flag, warm up, measure degraded, assert the ratio with both raw numbers in the failure message, then let the `chaos` fixture reset to clean in teardown. 6.4 (missing index) is not a chaos flag -- it's `make drop-index`/`fix-index`. 6.10 and 6.12 carry `slow` in addition to `pathology` (soak-ish; see tests/README.md's known-flaky list). """ import asyncio import random import statistics import subprocess import time from datetime import datetime, timezone import httpx import pytest from harness.config import ( GATEWAY_URL, POSTGRES_DSN, REPO_ROOT, SANCTIONS_LOOKUP_LOG_PATTERN, HEALTHZ_LOG_PATTERN, STREAM_NAME, ) from harness.containers import count_log_lines, rss_mb from harness.load import drive, measure_median from harness.payloads import tx_payload from harness.pg import explain, plan_node_types, statement_count from harness.streams import group_lag, wait_for_drain pytestmark = pytest.mark.pathology INSIGHTS_AGGREGATION_QUERY = """ SELECT category_id, count(*), sum(amount_minor) FROM transactions WHERE account_id = %s AND created_at >= now() - interval '120 days' GROUP BY category_id """ def _make(target, timeout=60): subprocess.run(["make", target], cwd=REPO_ROOT, check=True, timeout=timeout) # --------------------------------------------------------------------------- # 6.1 slow_sanctions # --------------------------------------------------------------------------- async def test_slow_sanctions_degrades_tail_latency(chaos, manifest, fresh_balances): """G6 §6.1: degraded p99/p50 >= 6 (baseline <= 4); degraded p99 >= 5x baseline p99.""" accounts = manifest["cold_account_ids_sample"][:20] for acc in accounts: fresh_balances(acc) def url_factory(): account_id = random.choice(accounts) return (f"{GATEWAY_URL}/api/transactions", dict(json=tx_payload(account_id, amount_minor=100))) async def run(): return await drive("POST", url_factory, total=60, concurrency=10, warmup=10) await chaos.set_profile("clean") baseline = await measure_median(3, run, key=lambda r: r.p95) await chaos.set_flag("slow_sanctions", True) degraded = await measure_median(3, run, key=lambda r: r.p95) baseline_ratio = baseline.p99 / baseline.p50 if baseline.p50 else float("inf") degraded_ratio = degraded.p99 / degraded.p50 if degraded.p50 else float("inf") assert baseline_ratio <= 4, f"baseline p99/p50={baseline_ratio:.2f} ({baseline})" assert degraded_ratio >= 6, f"degraded p99/p50={degraded_ratio:.2f} ({degraded})" assert degraded.p99 >= 5 * baseline.p99, ( f"degraded p99 not >= 5x baseline p99\nbaseline={baseline}\ndegraded={degraded}" ) # --------------------------------------------------------------------------- # 6.2 sanctions_errors # --------------------------------------------------------------------------- async def test_sanctions_errors_produces_bounded_error_rate(chaos, manifest, fresh_balances): """G6 §6.2: degraded 502/504 rate in [1%, 10%] over 300 requests; baseline == 0.""" accounts = manifest["cold_account_ids_sample"][:20] for acc in accounts: fresh_balances(acc) def url_factory(): account_id = random.choice(accounts) return (f"{GATEWAY_URL}/api/transactions", dict(json=tx_payload(account_id, amount_minor=100))) await chaos.set_profile("clean") baseline = await drive("POST", url_factory, total=300, concurrency=20, warmup=20) await chaos.set_flag("sanctions_errors", True) degraded = await drive("POST", url_factory, total=300, concurrency=20, warmup=20) gateway_errors = sum(n for status, n in degraded.status_counts.items() if status in (502, 504)) degraded_rate = gateway_errors / degraded.count assert baseline.error_rate == 0, f"baseline had errors: {baseline}" assert 0.01 <= degraded_rate <= 0.10, ( f"502/504 rate {degraded_rate:.1%} not in [1%,10%]\nbaseline={baseline}\ndegraded={degraded}" ) # --------------------------------------------------------------------------- # 6.3 n_plus_one # --------------------------------------------------------------------------- async def test_n_plus_one_multiplies_statement_count(chaos, manifest): """G6 §6.3: degraded issues >= 50 statements for one insights request (baseline <= 10); latency ratio >= 2.""" account_id = manifest["hot_account_ids"][0] async def one_request(): async with httpx.AsyncClient(timeout=15) as client: t0 = time.perf_counter() resp = await client.get(f"{GATEWAY_URL}/api/accounts/{account_id}/insights") elapsed = time.perf_counter() - t0 assert resp.status_code == 200 return elapsed await chaos.set_profile("clean") with statement_count(POSTGRES_DSN) as baseline_calls: baseline_latency = await one_request() await chaos.set_flag("n_plus_one", True) with statement_count(POSTGRES_DSN) as degraded_calls: degraded_latency = await one_request() assert degraded_calls.calls >= 50, f"degraded issued {degraded_calls.calls} statements" assert baseline_calls.calls <= 10, f"baseline issued {baseline_calls.calls} statements" ratio = degraded_latency / baseline_latency if baseline_latency else float("inf") assert ratio >= 2, ( f"latency ratio {ratio:.2f} not >= 2 " f"(baseline={baseline_latency*1000:.1f}ms, degraded={degraded_latency*1000:.1f}ms)" ) # --------------------------------------------------------------------------- # 6.4 missing index (make drop-index / fix-index, not a chaos flag) # --------------------------------------------------------------------------- def test_missing_index_forces_seq_scan(manifest): """G6 §6.4: EXPLAIN (ANALYZE, FORMAT JSON) on the insights aggregation shows a Seq Scan without the index and an Index/Bitmap Scan with it; unindexed latency >= 3x indexed. Teardown restores the dropped state.""" account_id = manifest["hot_account_ids"][0] _make("drop-index") plan = explain(POSTGRES_DSN, INSIGHTS_AGGREGATION_QUERY, (account_id,)) unindexed_types = plan_node_types(plan) unindexed_time = plan["Plan"]["Actual Total Time"] assert "Seq Scan" in unindexed_types, unindexed_types try: _make("fix-index") plan = explain(POSTGRES_DSN, INSIGHTS_AGGREGATION_QUERY, (account_id,)) indexed_types = plan_node_types(plan) indexed_time = plan["Plan"]["Actual Total Time"] assert indexed_types & {"Index Scan", "Bitmap Index Scan", "Bitmap Heap Scan"}, indexed_types assert unindexed_time >= 3 * indexed_time, ( f"unindexed {unindexed_time:.2f}ms not >= 3x indexed {indexed_time:.2f}ms" ) finally: _make("drop-index") # --------------------------------------------------------------------------- # 6.5 pool_starvation -- no automated test. # # test_pool_starvation_plateaus_throughput (removed) asserted baseline # throughput scales >= 2.5x from concurrency=5 to concurrency=25. Isolated # reproduction (a bare GET /healthz with zero backend work, measured from # inside the gateway container, bypassing Docker's networking entirely) # showed the same flat-to-negative scaling regardless of any chaos flag: # ~600rps at c5 vs ~480-530rps at c25. This is uvicorn running a single # worker per service, exactly as phase-0-base-project-spec.md §3 mandates # ("one worker per service") -- there is no second process/core for that one # service to spread load across, so throughput cannot scale with added # concurrency in this deployment regardless of pool_starvation. The # assertion conflicted with the spec's own mandated architecture, not with # this implementation; removed by user decision rather than loosened. # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # 6.6 row_lock_contention # --------------------------------------------------------------------------- async def test_row_lock_contention_on_hot_account(chaos, manifest, fresh_balances): """G6 §6.6: with the flag on, 25 concurrent transactions against one hot account has p95 >= 3x the same burst spread across 25 cold accounts; with the flag off, that ratio is <= 1.8.""" hot_account = manifest["hot_account_ids"][0] cold_accounts = manifest["cold_account_ids_sample"][:25] fresh_balances(hot_account) for acc in cold_accounts: fresh_balances(acc) def hot_factory(): return (f"{GATEWAY_URL}/api/transactions", dict(json=tx_payload(hot_account, amount_minor=100))) def spread_factory(): account_id = random.choice(cold_accounts) return (f"{GATEWAY_URL}/api/transactions", dict(json=tx_payload(account_id, amount_minor=100))) await chaos.set_profile("clean") off_hot = await drive("POST", hot_factory, total=25, concurrency=25, warmup=5) off_spread = await drive("POST", spread_factory, total=25, concurrency=25, warmup=5) off_ratio = off_hot.p95 / off_spread.p95 if off_spread.p95 else float("inf") await chaos.set_flag("row_lock_contention", True) on_hot = await drive("POST", hot_factory, total=25, concurrency=25, warmup=5) on_spread = await drive("POST", spread_factory, total=25, concurrency=25, warmup=5) on_ratio = on_hot.p95 / on_spread.p95 if on_spread.p95 else float("inf") assert on_ratio >= 3, f"flag on: hot={on_hot} spread={on_spread} ratio={on_ratio:.2f}" assert off_ratio <= 1.8, f"flag off: hot={off_hot} spread={off_spread} ratio={off_ratio:.2f}" # --------------------------------------------------------------------------- # 6.7 blocking_event_loop # --------------------------------------------------------------------------- async def test_blocking_event_loop_serializes_concurrent_requests(chaos, manifest, fresh_balances): """G6 §6.7: wall-clock for 10 concurrent requests / wall-clock for 1 request. Degraded ratio >= 6 (serialised); baseline ratio <= 2.5 (concurrent).""" accounts = manifest["cold_account_ids_sample"][:10] for acc in accounts: fresh_balances(acc) async def one_request(account_id): async with httpx.AsyncClient(timeout=15) as client: resp = await client.post( f"{GATEWAY_URL}/api/transactions", json=tx_payload(account_id, amount_minor=100) ) assert resp.status_code == 201 async def single_wall_clock(): t0 = time.perf_counter() await one_request(accounts[0]) return time.perf_counter() - t0 async def concurrent_wall_clock(): t0 = time.perf_counter() await asyncio.gather(*[one_request(acc) for acc in accounts]) return time.perf_counter() - t0 await chaos.set_profile("clean") baseline_single = await single_wall_clock() baseline_concurrent = await concurrent_wall_clock() baseline_ratio = baseline_concurrent / baseline_single await chaos.set_flag("blocking_event_loop", True) degraded_single = await single_wall_clock() degraded_concurrent = await concurrent_wall_clock() degraded_ratio = degraded_concurrent / degraded_single assert degraded_ratio >= 6, ( f"degraded: 10-concurrent {degraded_concurrent:.2f}s vs 1-request " f"{degraded_single:.2f}s ratio={degraded_ratio:.2f}" ) assert baseline_ratio <= 2.5, ( f"baseline: 10-concurrent {baseline_concurrent:.2f}s vs 1-request " f"{baseline_single:.2f}s ratio={baseline_ratio:.2f}" ) # --------------------------------------------------------------------------- # 6.8 retry_storm (with sanctions_errors also on) # --------------------------------------------------------------------------- async def test_retry_storm_multiplies_sanctions_lookups(chaos, manifest, fresh_balances): """G6 §6.8: with sanctions_errors also on, degraded >= 2.0 sanctions lookup log lines per transaction; baseline (retry off, errors on) <= 1.2.""" account_id = manifest["clean_account_id"] fresh_balances(account_id) n = 20 async def burst(): since = datetime.now(timezone.utc) async with httpx.AsyncClient(timeout=15) as client: for _ in range(n): await client.post( f"{GATEWAY_URL}/api/transactions", json=tx_payload(account_id, amount_minor=100) ) return count_log_lines("sanctions", SANCTIONS_LOOKUP_LOG_PATTERN, since) await chaos.set_profile("clean") await chaos.set_flag("sanctions_errors", True) baseline_lookups = await burst() baseline_ratio = baseline_lookups / n await chaos.set_flag("retry_storm", True) degraded_lookups = await burst() degraded_ratio = degraded_lookups / n assert degraded_ratio >= 2.0, f"degraded {degraded_ratio:.2f} lookups/tx ({degraded_lookups}/{n})" assert baseline_ratio <= 1.2, f"baseline {baseline_ratio:.2f} lookups/tx ({baseline_lookups}/{n})" # --------------------------------------------------------------------------- # 6.9 no_batching (sanctions_errors off) # --------------------------------------------------------------------------- async def test_no_batching_multiplies_sanctions_lookups(chaos, manifest, fresh_balances): """G6 §6.9: sanctions_errors off. Degraded >= 2.0 lookups/tx (baseline <= 1.2); latency ratio >= 1.5.""" account_id = manifest["clean_account_id"] fresh_balances(account_id) n = 20 async def burst(): since = datetime.now(timezone.utc) t0 = time.perf_counter() async with httpx.AsyncClient(timeout=15) as client: for _ in range(n): await client.post( f"{GATEWAY_URL}/api/transactions", json=tx_payload(account_id, amount_minor=100) ) elapsed = time.perf_counter() - t0 return count_log_lines("sanctions", SANCTIONS_LOOKUP_LOG_PATTERN, since), elapsed await chaos.set_profile("clean") baseline_lookups, baseline_elapsed = await burst() baseline_ratio = baseline_lookups / n await chaos.set_flag("no_batching", True) degraded_lookups, degraded_elapsed = await burst() degraded_ratio = degraded_lookups / n assert degraded_ratio >= 2.0, f"degraded {degraded_ratio:.2f} lookups/tx" assert baseline_ratio <= 1.2, f"baseline {baseline_ratio:.2f} lookups/tx" latency_ratio = (degraded_elapsed / n) / (baseline_elapsed / n) assert latency_ratio >= 1.5, ( f"latency ratio {latency_ratio:.2f} not >= 1.5 " f"(baseline {baseline_elapsed/n*1000:.1f}ms/tx, degraded {degraded_elapsed/n*1000:.1f}ms/tx)" ) # --------------------------------------------------------------------------- # 6.10 unbounded_cache (slow) # --------------------------------------------------------------------------- @pytest.mark.slow async def test_unbounded_cache_leaks_memory(chaos, manifest): """G6 §6.10: RSS growth after 2000 requests with distinct cache keys (the pathology keys on (account_id, timestamp_ms), so plain repeated calls already miss every time), sampled 30s after the burst. Degraded growth >= 20MB and still elevated at +30s; baseline growth <= 8MB.""" account_id = manifest["hot_account_ids"][0] async def burst_distinct(n): async with httpx.AsyncClient(timeout=15) as client: sem = asyncio.Semaphore(20) async def one(): async with sem: await client.get(f"{GATEWAY_URL}/api/accounts/{account_id}/insights") await asyncio.gather(*[one() for _ in range(n)]) await chaos.set_profile("clean") before = rss_mb("insights") await burst_distinct(2000) await asyncio.sleep(30) baseline_growth = rss_mb("insights") - before await chaos.set_flag("unbounded_cache", True) before2 = rss_mb("insights") await burst_distinct(2000) await asyncio.sleep(30) degraded_growth = rss_mb("insights") - before2 assert baseline_growth <= 8, f"baseline RSS grew {baseline_growth:.1f}MB" assert degraded_growth >= 20, f"degraded RSS grew {degraded_growth:.1f}MB" # --------------------------------------------------------------------------- # 6.11 slow_consumer # --------------------------------------------------------------------------- async def test_slow_consumer_builds_lag(chaos, manifest, fresh_balances, redis_client): """G6 §6.11: after a 150-transaction burst, degraded notifier lag >= 30 at burst end and still > 0 after 10s; baseline drains to 0 within 10s.""" account_id = manifest["clean_account_id"] fresh_balances(account_id) async def burst(n): async with httpx.AsyncClient(timeout=15) as client: sem = asyncio.Semaphore(20) async def one(): async with sem: await client.post( f"{GATEWAY_URL}/api/transactions", json=tx_payload(account_id, amount_minor=100) ) await asyncio.gather(*[one() for _ in range(n)]) await chaos.set_profile("clean") await burst(150) wait_for_drain(redis_client, STREAM_NAME, "notifier", timeout=10) await chaos.set_flag("slow_consumer", True) await burst(150) lag_at_end = group_lag(redis_client, STREAM_NAME, "notifier").get("lag") or 0 assert lag_at_end >= 30, f"lag at burst end was {lag_at_end}, want >= 30" await asyncio.sleep(10) lag_after = group_lag(redis_client, STREAM_NAME, "notifier").get("lag") or 0 assert lag_after > 0, f"expected lag still > 0 after 10s, got {lag_after}" # --------------------------------------------------------------------------- # 6.12 cold_cache (slow) # --------------------------------------------------------------------------- @pytest.mark.slow async def test_cold_cache_produces_periodic_spikes(chaos, manifest): """G6 §6.12: same insights request at 1rps for 150s; count samples > 3x the median. Degraded >= 2 spikes (a 60s TTL over 150s); baseline <= 1.""" account_id = manifest["hot_account_ids"][0] async def sample_for_150s(): latencies = [] async with httpx.AsyncClient(timeout=15) as client: for _ in range(150): t0 = time.perf_counter() await client.get(f"{GATEWAY_URL}/api/accounts/{account_id}/insights") latencies.append((time.perf_counter() - t0) * 1000) await asyncio.sleep(1) return latencies def count_spikes(latencies): median = statistics.median(latencies) return sum(1 for l in latencies if l > 3 * median) await chaos.set_profile("clean") baseline_latencies = await sample_for_150s() baseline_spikes = count_spikes(baseline_latencies) await chaos.set_flag("cold_cache", True) degraded_latencies = await sample_for_150s() degraded_spikes = count_spikes(degraded_latencies) assert baseline_spikes <= 1, f"baseline had {baseline_spikes} spikes: {baseline_latencies}" assert degraded_spikes >= 2, f"degraded had {degraded_spikes} spikes: {degraded_latencies}" # --------------------------------------------------------------------------- # 6.13 poison_data # --------------------------------------------------------------------------- async def test_poison_data_causes_unhandled_exceptions(chaos, manifest): """G6 §6.13: requests against poison accounts produce >= 1 HTTP 500 when the flag is on; baseline produces zero 500s (4xx or 200/201 only).""" poison_accounts = manifest["poison_account_ids"] async def probe(): statuses = [] async with httpx.AsyncClient(timeout=15) as client: for account_id in poison_accounts: resp = await client.post( f"{GATEWAY_URL}/api/transactions", json=tx_payload(account_id, amount_minor=100) ) statuses.append(resp.status_code) return statuses await chaos.set_profile("clean") baseline_statuses = await probe() assert all(s in (200, 201) or 400 <= s < 500 for s in baseline_statuses), baseline_statuses await chaos.set_flag("poison_data", True) degraded_statuses = await probe() assert 500 in degraded_statuses, degraded_statuses # --------------------------------------------------------------------------- # 6.14 chatty_health # --------------------------------------------------------------------------- async def test_chatty_health_polls_downstreams(chaos, manifest, fresh_balances): """G6 §6.14: on a single transaction, degraded produces >= 5 healthz log lines across downstream services; baseline == 0.""" account_id = manifest["clean_account_id"] fresh_balances(account_id) downstreams = ["ledger", "fraud", "sanctions", "insights", "notifier"] async def one_transaction_healthz_hits(): since = datetime.now(timezone.utc) async with httpx.AsyncClient(timeout=15) as client: await client.post( f"{GATEWAY_URL}/api/transactions", json=tx_payload(account_id, amount_minor=100) ) return sum(count_log_lines(svc, HEALTHZ_LOG_PATTERN, since) for svc in downstreams) await chaos.set_profile("clean") baseline_hits = await one_transaction_healthz_hits() assert baseline_hits == 0, f"baseline had {baseline_hits} healthz hits: expected 0" await chaos.set_flag("chatty_health", True) degraded_hits = await one_transaction_healthz_hits() assert degraded_hits >= 5, f"degraded had {degraded_hits} healthz hits: expected >= 5"