LedgerLine/tests/test_pathologies.py

"""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"