## Invariants
- No job is claimed by more than one worker.
- Claimed jobs move from READY to CLAIMED exactly once.
- Failed claims do not modify job state.
- Total claimed count equals distinct worker claims.
- No transaction remains open after the benchmark.

-- No duplicate claims.
SELECT job_id, count(*)
FROM bench.job_claim_log
GROUP BY job_id
HAVING count(*) > 1;

-- Status/count reconciliation.
SELECT status, count(*)
FROM bench.job
GROUP BY status
ORDER BY status;

-- Claimed jobs have a log.
SELECT count(*)
FROM bench.job j
WHERE j.status = 'CLAIMED'
  AND NOT EXISTS (
    SELECT 1
    FROM bench.job_claim_log l
    WHERE l.job_id = j.job_id
  );

pgbench \
  --client=16 \
  --jobs=4 \
  --time=120 \
  --progress=10 \
  --file=bench/claim_job.sql \
  appdb

CREATE SCHEMA IF NOT EXISTS bench;

DROP TABLE IF EXISTS bench.job_claim_log;
DROP TABLE IF EXISTS bench.job;

CREATE TABLE bench.job (
  job_id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  tenant_id bigint NOT NULL,
  status text NOT NULL CHECK (status IN ('READY', 'CLAIMED', 'DONE', 'FAILED')),
  priority integer NOT NULL DEFAULT 0,
  available_at timestamptz NOT NULL DEFAULT clock_timestamp(),
  claimed_by text,
  claimed_at timestamptz,
  payload jsonb NOT NULL DEFAULT '{}'::jsonb
);

CREATE INDEX bench_job_ready_idx
ON bench.job (priority DESC, available_at, job_id)
WHERE status = 'READY';

CREATE TABLE bench.job_claim_log (
  claim_id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  job_id bigint NOT NULL REFERENCES bench.job(job_id),
  worker_id text NOT NULL,
  claimed_at timestamptz NOT NULL DEFAULT clock_timestamp(),
  UNIQUE (job_id)
);

INSERT INTO bench.job (tenant_id, status, priority, available_at, payload)
SELECT
  CASE
    WHEN gs <= 800000 THEN 1
    WHEN gs <= 950000 THEN 2
    ELSE 3
  END AS tenant_id,
  'READY',
  (random() * 100)::integer,
  clock_timestamp() - ((random() * 86400)::integer || ' seconds')::interval,
  jsonb_build_object('source', 'benchmark', 'n', gs)
FROM generate_series(1, 1000000) AS gs;

ANALYZE bench.job;

CREATE OR REPLACE FUNCTION bench.claim_next_job(p_worker_id text)
RETURNS bigint
LANGUAGE plpgsql
AS $$
DECLARE
  v_job_id bigint;
BEGIN
  SELECT j.job_id
  INTO v_job_id
  FROM bench.job j
  WHERE j.status = 'READY'
    AND j.available_at <= clock_timestamp()
  ORDER BY j.priority DESC, j.available_at, j.job_id
  FOR UPDATE SKIP LOCKED
  LIMIT 1;

  IF v_job_id IS NULL THEN
    RETURN NULL;
  END IF;

  UPDATE bench.job j
  SET status = 'CLAIMED',
      claimed_by = p_worker_id,
      claimed_at = clock_timestamp()
  WHERE j.job_id = v_job_id;

  INSERT INTO bench.job_claim_log (job_id, worker_id)
  VALUES (v_job_id, p_worker_id);

  RETURN v_job_id;
END;
$$;

\set worker_id random(1, 1000000)
SELECT bench.claim_next_job('worker-' || :worker_id);

for c in 1 4 8 16 32 64; do
  pgbench \
    --client=$c \
    --jobs=4 \
    --time=120 \
    --progress=10 \
    --file=bench/claim_job.sql \
    appdb \
    > results/claim_job_c${c}.txt
 done

TRUNCATE bench.job_claim_log;
UPDATE bench.job
SET status = 'READY',
    claimed_by = NULL,
    claimed_at = NULL;
VACUUM ANALYZE bench.job;

SELECT pg_stat_reset();
SELECT pg_stat_statements_reset();

SELECT
  calls,
  total_exec_time,
  mean_exec_time,
  max_exec_time,
  rows,
  shared_blks_hit,
  shared_blks_read,
  temp_blks_written,
  wal_records,
  wal_bytes,
  left(query, 160) AS query_sample
FROM pg_stat_statements
ORDER BY total_exec_time DESC
LIMIT 20;

SELECT
  schemaname,
  funcname,
  calls,
  total_time,
  self_time,
  total_time / NULLIF(calls, 0) AS avg_total_ms,
  self_time / NULLIF(calls, 0) AS avg_self_ms
FROM pg_stat_user_functions
WHERE schemaname = 'bench'
ORDER BY total_time DESC;

SELECT
  wait_event_type,
  wait_event,
  count(*)
FROM pg_stat_activity
WHERE datname = current_database()
GROUP BY wait_event_type, wait_event
ORDER BY count(*) DESC;

CREATE TABLE bench.case_file (
  case_id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  tenant_id bigint NOT NULL,
  status text NOT NULL,
  created_at timestamptz NOT NULL DEFAULT clock_timestamp()
);

CREATE INDEX bench_case_file_tenant_status_idx
ON bench.case_file (tenant_id, status, created_at DESC);

INSERT INTO bench.case_file (tenant_id, status, created_at)
SELECT
  CASE WHEN gs <= 900000 THEN 1 ELSE gs END,
  CASE WHEN random() < 0.9 THEN 'OPEN' ELSE 'CLOSED' END,
  clock_timestamp() - ((random() * 31536000)::bigint || ' seconds')::interval
FROM generate_series(1, 1000000) gs;

ANALYZE bench.case_file;

CREATE OR REPLACE FUNCTION bench.count_open_cases_static(p_tenant_id bigint)
RETURNS bigint
LANGUAGE plpgsql
AS $$
DECLARE
  v_count bigint;
BEGIN
  SELECT count(*)
  INTO v_count
  FROM bench.case_file c
  WHERE c.tenant_id = p_tenant_id
    AND c.status = 'OPEN';

  RETURN v_count;
END;
$$;

CREATE OR REPLACE FUNCTION bench.count_open_cases_dynamic(p_tenant_id bigint)
RETURNS bigint
LANGUAGE plpgsql
AS $$
DECLARE
  v_count bigint;
BEGIN
  EXECUTE
    'SELECT count(*) FROM bench.case_file c WHERE c.tenant_id = $1 AND c.status = $2'
  INTO v_count
  USING p_tenant_id, 'OPEN';

  RETURN v_count;
END;
$$;

\set tenant random(1, 1000000)
SELECT bench.count_open_cases_static(
  CASE WHEN :tenant <= 900000 THEN 1 ELSE :tenant END
);

\set case_id random(1, 1000000)
SELECT app.transition_case(:case_id, 'ESCALATED', 'benchmark');

WITH candidate AS (
  SELECT case_id
  FROM app.case_file
  WHERE status = 'OPEN'
  ORDER BY random()
  LIMIT 1
)
SELECT app.transition_case(candidate.case_id, 'ESCALATED', 'benchmark')
FROM candidate;

CREATE TABLE bench.transition_candidate AS
SELECT case_id, row_number() OVER () AS rn
FROM app.case_file
WHERE status = 'OPEN';

CREATE UNIQUE INDEX ON bench.transition_candidate (rn);
ANALYZE bench.transition_candidate;

\set rn random(1, 1000000)
SELECT app.transition_case(c.case_id, 'ESCALATED', 'benchmark')
FROM bench.transition_candidate c
WHERE c.rn = :rn;

SELECT status, count(*)
FROM app.case_file
GROUP BY status;

SELECT count(*) FROM app.audit_event WHERE reason_code = 'benchmark';
SELECT count(*) FROM app.outbox_event WHERE aggregate_type = 'case_file';

CREATE TABLE bench.case_group_counter (
  group_id bigint PRIMARY KEY,
  open_count bigint NOT NULL
);

CREATE TABLE bench.case_item (
  case_id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  group_id bigint NOT NULL,
  status text NOT NULL
);

CREATE OR REPLACE FUNCTION bench.close_case_with_counter(p_case_id bigint)
RETURNS void
LANGUAGE plpgsql
AS $$
DECLARE
  v_group_id bigint;
BEGIN
  SELECT group_id
  INTO STRICT v_group_id
  FROM bench.case_item
  WHERE case_id = p_case_id
  FOR UPDATE;

  UPDATE bench.case_item
  SET status = 'CLOSED'
  WHERE case_id = p_case_id
    AND status = 'OPEN';

  IF FOUND THEN
    UPDATE bench.case_group_counter
    SET open_count = open_count - 1
    WHERE group_id = v_group_id;
  END IF;
END;
$$;

\set case_id random(1, 10000)
SELECT bench.close_case_with_counter(:case_id);

SELECT g.group_id, g.open_count, actual.actual_open_count
FROM bench.case_group_counter g
JOIN (
  SELECT group_id, count(*) AS actual_open_count
  FROM bench.case_item
  WHERE status = 'OPEN'
  GROUP BY group_id
) actual USING (group_id)
WHERE g.open_count <> actual.actual_open_count;

## Benchmark Run
- Date: 2026-07-03
- PostgreSQL: 18.x
- Dataset: 10M case_file rows, 100M audit rows
- Script: bench/transition_case.sql
- Clients: 1, 8, 32, 64
- Duration: 180s + 30s warmup
- Hardware: 8 vCPU, 32GB RAM, NVMe
- Config changes: shared_buffers=8GB, work_mem=64MB
- App version: git sha ...
- Migration version: ...

pgbench \
  --client=32 \
  --jobs=8 \
  --time=300 \
  --rate=500 \
  --latency-limit=200 \
  --file=bench/transition_case.sql \
  appdb

\set case_id random(1, 1000000)
SELECT app.transition_case(:case_id, 'IMPOSSIBLE_STATUS', 'benchmark_failure');

CREATE OR REPLACE FUNCTION bench.try_invalid_transition(p_case_id bigint)
RETURNS text
LANGUAGE plpgsql
AS $$
BEGIN
  PERFORM app.transition_case(p_case_id, 'IMPOSSIBLE_STATUS', 'benchmark_failure');
  RETURN 'unexpected_success';
EXCEPTION
  WHEN SQLSTATE 'P2401' THEN
    RETURN 'expected_domain_failure';
END;
$$;

\set case_id random(1, 1000000)
SELECT bench.try_invalid_transition(:case_id);

CREATE TABLE bench.retry_observation (
  observation_id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  worker_id text NOT NULL,
  attempt_no integer NOT NULL,
  sqlstate text,
  success boolean NOT NULL,
  observed_at timestamptz NOT NULL DEFAULT clock_timestamp()
);

BEGIN;
EXPLAIN (ANALYZE, BUFFERS, WAL)
UPDATE app.case_file
SET status = 'ESCALATED'
WHERE status = 'OPEN'
  AND tenant_id = 42
LIMIT 0; -- invalid syntax: use controlled predicate instead
ROLLBACK;

BEGIN;
WITH candidate AS (
  SELECT case_id
  FROM app.case_file
  WHERE status = 'OPEN'
    AND tenant_id = 42
  ORDER BY case_id
  LIMIT 1000
)
EXPLAIN (ANALYZE, BUFFERS, WAL)
UPDATE app.case_file c
SET status = 'ESCALATED'
FROM candidate x
WHERE x.case_id = c.case_id;
ROLLBACK;

SELECT count(*) FROM app.get_large_report(42);

-- Full result cost.
SELECT count(*) FROM app.get_large_report(42);

-- Page-like access.
SELECT * FROM app.get_large_report_page(42, 1000, NULL);

-- Query plan for underlying statement.
EXPLAIN (ANALYZE, BUFFERS)
SELECT ... underlying report query ...;

routine: app.transition_case
workload: bench/transition_case.sql
dataset: case-management-10m-v3
thresholds:
  mean_latency_ms_max: 25
  p95_latency_ms_max: 80
  p99_latency_ms_max: 200
  tps_min: 500
  error_rate_max: 0.001
  duplicate_side_effects_max: 0
  wal_bytes_per_tx_max: 5000
  lock_wait_ratio_max: 0.02

CREATE TABLE bench.benchmark_run (
  run_id uuid PRIMARY KEY,
  benchmark_name text NOT NULL,
  git_sha text,
  migration_version text,
  postgres_version text NOT NULL DEFAULT version(),
  dataset_name text NOT NULL,
  clients integer NOT NULL,
  duration_seconds integer NOT NULL,
  started_at timestamptz NOT NULL DEFAULT clock_timestamp(),
  notes text
);

CREATE TABLE bench.benchmark_metric (
  run_id uuid NOT NULL REFERENCES bench.benchmark_run(run_id),
  metric_name text NOT NULL,
  metric_value numeric NOT NULL,
  unit text NOT NULL,
  PRIMARY KEY (run_id, metric_name)
);

SELECT
  r.started_at,
  r.git_sha,
  r.clients,
  m.metric_value AS p95_latency_ms
FROM bench.benchmark_run r
JOIN bench.benchmark_metric m USING (run_id)
WHERE r.benchmark_name = 'transition_case'
  AND m.metric_name = 'p95_latency_ms'
ORDER BY r.started_at;

# Benchmark Report: app.transition_case

## Question
Can `app.transition_case()` sustain expected peak traffic of 500 transitions/sec with p95 < 80ms and no duplicate audit/outbox side effects?

## Workload
- Dataset: 10M cases, 50M audit rows, 10M outbox rows
- Distribution: tenant 1 owns 70% of open cases
- Clients: 1, 8, 16, 32, 64
- Duration: 5 minutes each, 30s warmup discarded
- Script: bench/transition_case.sql

## Correctness Invariants
- no invalid state transition accepted
- every successful transition has exactly one audit event
- every successful transition has exactly one outbox event
- no case has duplicate transition for same command id

## Results
| Clients | TPS | p95 ms | p99 ms | Error % | Notes |
|---:|---:|---:|---:|---:|---|
| 1 | ... | ... | ... | ... | baseline |
| 8 | ... | ... | ... | ... | stable |
| 16 | ... | ... | ... | ... | stable |
| 32 | ... | ... | ... | ... | lock wait begins |
| 64 | ... | ... | ... | ... | saturation |

## Diagnosis
Saturation begins around 32 clients due to row lock contention on hot tenant transition sequence.

## Decision
Accept for expected load with rate limit. For batch escalation, use partitioned worker by tenant group.