How We Fixed a PostgreSQL 16 Deadlock Issue That Caused 5 Production Outages in 2 Weeks

\\nimport psycopg2\\nimport threading\\nimport time\\nimport logging\\nfrom typing import List, Dict\\n\\n# Configure logging to capture deadlock errors\\nlogging.basicConfig(\\n    level=logging.INFO,\\n    format=\"%(asctime)s - %(threadName)s - %(message)s\"\\n)\\nlogger = logging.getLogger(__name__)\\n\\n# Database connection parameters - update these to match your PG16 instance\\nDB_PARAMS = {\\n    \"host\": \"localhost\",\\n    \"port\": 5432,\\n    \"database\": \"deadlock_test\",\\n    \"user\": \"postgres\",\\n    \"password\": \"postgres\"\\n}\\n\\ndef create_test_table(cur):\\n    \"\"\"Create the test table with the same schema that triggered our production deadlock\"\"\"\\n    try:\\n        cur.execute(\"\"\"\\n            CREATE TABLE IF NOT EXISTS transaction_logs (\\n                id BIGSERIAL PRIMARY KEY,\\n                user_id BIGINT NOT NULL,\\n                amount NUMERIC(19,4) NOT NULL,\\n                status VARCHAR(20) NOT NULL DEFAULT 'pending',\\n                created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),\\n                processed_at TIMESTAMPTZ\\n            )\\n        \"\"\")\\n        # Insert 1M rows to trigger parallel sequential scans in PG16\\n        cur.execute(\"SELECT COUNT(*) FROM transaction_logs\")\\n        if cur.fetchone()[0] == 0:\\n            logger.info(\"Seeding 1M test rows...\")\\n            cur.execute(\"\"\"\\n                INSERT INTO transaction_logs (user_id, amount, status)\\n                SELECT generate_series(1,1000000), (random() * 1000)::NUMERIC(19,4), 'pending'\\n            \"\"\")\\n        cur.execute(\"CREATE INDEX IF NOT EXISTS idx_transaction_logs_status ON transaction_logs(status)\")\\n        cur.execute(\"CREATE INDEX IF NOT EXISTS idx_transaction_logs_user_id ON transaction_logs(user_id)\")\\n        logger.info(\"Test table initialized successfully\")\\n    except Exception as e:\\n        logger.error(f\"Failed to create test table: {e}\")\\n        raise\\n\\ndef run_oltp_update(thread_id: int, iterations: int = 1000):\\n    \"\"\"Simulate OLTP update workload: row-level lock + update\"\"\"\\n    conn = None\\n    try:\\n        conn = psycopg2.connect(**DB_PARAMS)\\n        conn.autocommit = False\\n        cur = conn.cursor()\\n        for i in range(iterations):\\n            try:\\n                # Select a random pending transaction and lock it\\n                cur.execute(\"\"\"\\n                    SELECT id FROM transaction_logs \\n                    WHERE status = 'pending' \\n                    ORDER BY RANDOM() \\n                    LIMIT 1 \\n                    FOR UPDATE\\n                \"\"\")\\n                row = cur.fetchone()\\n                if not row:\\n                    logger.warning(f\"Thread {thread_id}: No pending rows found\")\\n                    continue\\n                txn_id = row[0]\\n                # Update the locked row\\n                cur.execute(\"\"\"\\n                    UPDATE transaction_logs \\n                    SET status = 'processed', processed_at = NOW() \\n                    WHERE id = %s\\n                \"\"\", (txn_id,))\\n                conn.commit()\\n                if i % 100 == 0:\\n                    logger.info(f\"Thread {thread_id}: Completed {i} OLTP updates\")\\n            except psycopg2.Error as e:\\n                if \"deadlock detected\" in str(e).lower():\\n                    logger.error(f\"Thread {thread_id}: Deadlock detected! {e}\")\\n                    conn.rollback()\\n                else:\\n                    logger.error(f\"Thread {thread_id}: Database error: {e}\")\\n                    conn.rollback()\\n    except Exception as e:\\n        logger.error(f\"Thread {thread_id}: Fatal error: {e}\")\\n    finally:\\n        if conn:\\n            conn.close()\\n\\ndef run_olap_scan(thread_id: int, iterations: int = 100):\\n    \"\"\"Simulate OLAP scan workload: parallel sequential scan with PG16's PSS feature\"\"\"\\n    conn = None\\n    try:\\n        conn = psycopg2.connect(**DB_PARAMS)\\n        # Enable parallel queries explicitly (PG16 default is on, but we force it for reproducibility)\\n        conn.autocommit = True\\n        cur = conn.cursor()\\n        # Force parallel sequential scan by setting low parallel cost\\n        cur.execute(\"SET max_parallel_workers_per_gather = 4\")\\n        cur.execute(\"SET parallel_setup_cost = 0\")\\n        cur.execute(\"SET parallel_tuple_cost = 0\")\\n        for i in range(iterations):\\n            try:\\n                # This query triggers parallel sequential scan in PG16 for large tables\\n                cur.execute(\"\"\"\\n                    SELECT status, COUNT(*), SUM(amount) \\n                    FROM transaction_logs \\n                    WHERE created_at > NOW() - INTERVAL '24 hours'\\n                    GROUP BY status\\n                \"\"\")\\n                result = cur.fetchall()\\n                if i % 10 == 0:\\n                    logger.info(f\"Thread {thread_id}: Completed OLAP scan, {len(result)} groups\")\\n            except psycopg2.Error as e:\\n                logger.error(f\"Thread {thread_id}: OLAP scan error: {e}\")\\n    except Exception as e:\\n        logger.error(f\"Thread {thread_id}: Fatal error: {e}\")\\n    finally:\\n        if conn:\\n            conn.close()\\n\\nif __name__ == \"__main__\":\\n    # Initialize test table\\n    conn = psycopg2.connect(**DB_PARAMS)\\n    cur = conn.cursor()\\n    create_test_table(cur)\\n    conn.commit()\\n    conn.close()\\n\\n    # Start OLTP update threads (simulate production transaction volume)\\n    oltp_threads = []\\n    for i in range(8):\\n        t = threading.Thread(\\n            target=run_oltp_update,\\n            args=(i, 1000),\\n            name=f\"OLTP-Worker-{i}\"\\n        )\\n        oltp_threads.append(t)\\n        t.start()\\n\\n    # Start OLAP scan threads (simulate nightly reporting workload)\\n    olap_threads = []\\n    for i in range(4):\\n        t = threading.Thread(\\n            target=run_olap_scan,\\n            args=(i, 100),\\n            name=f\"OLAP-Worker-{i}\"\\n        )\\n        olap_threads.append(t)\\n        t.start()\\n\\n    # Wait for all threads to complete\\n    for t in oltp_threads + olap_threads:\\n        t.join()\\n\\n    logger.info(\"Reproduction run completed\")\\n

\\npackage main\\n\\nimport (\\n\\t\"bufio\"\\n\\t\"context\"\\n\\t\"database/sql\"\\n\\t\"fmt\"\\n\\t\"log\"\\n\\t\"os\"\\n\\t\"os/signal\"\\n\\t\"regexp\"\\n\\t\"strings\"\\n\\t\"syscall\"\\n\\t\"time\"\\n\\n\\t_ \"github.com/lib/pq\" // PostgreSQL driver\\n)\\n\\n// DBConfig holds database connection parameters\\ntype DBConfig struct {\\n\\tHost     string\\n\\tPort     int\\n\\tUser     string\\n\\tPassword string\\n\\tDBName   string\\n}\\n\\n// DeadlockDetector tails PostgreSQL logs and applies mitigations when deadlocks are detected\\ntype DeadlockDetector struct {\\n\\tdb       *sql.DB\\n\\tlogPath  string\\n\\tdeadlock *regexp.Regexp\\n\\tmitigated map[string]bool // Track which PIDs have been mitigated to avoid duplicates\\n}\\n\\n// NewDeadlockDetector initializes a new detector instance\\nfunc NewDeadlockDetector(cfg DBConfig, logPath string) (*DeadlockDetector, error) {\\n\\tconnStr := fmt.Sprintf(\\n\\t\\t\"host=%s port=%d user=%s password=%s dbname=%s sslmode=disable\",\\n\\t\\tcfg.Host, cfg.Port, cfg.User, cfg.Password, cfg.DBName,\\n\\t)\\n\\tdb, err := sql.Open(\"postgres\", connStr)\\n\\tif err != nil {\\n\\t\\treturn nil, fmt.Errorf(\"failed to connect to database: %w\", err)\\n\\t}\\n\\t// Verify connection\\n\\tif err := db.Ping(); err != nil {\\n\\t\\treturn nil, fmt.Errorf(\"failed to ping database: %w\", err)\\n\\t}\\n\\t// Regex to match PostgreSQL deadlock log entries (PG16 format)\\n\\tdeadlockRegex, err := regexp.Compile(`deadlock detected\\\\s+DETAIL:.*?process \\\\d+`)\\n\\tif err != nil {\\n\\t\\treturn nil, fmt.Errorf(\"failed to compile deadlock regex: %w\", err)\\n\\t}\\n\\treturn &DeadlockDetector{\\n\\t\\tdb:       db,\\n\\t\\tlogPath:  logPath,\\n\\t\\tdeadlock: deadlockRegex,\\n\\t\\tmitigated: make(map[string]bool),\\n\\t}, nil\\n}\\n\\n// Start begins tailing the log file and processing entries\\nfunc (d *DeadlockDetector) Start(ctx context.Context) error {\\n\\tfile, err := os.Open(d.logPath)\\n\\tif err != nil {\\n\\t\\treturn fmt.Errorf(\"failed to open log file %s: %w\", d.logPath, err)\\n\\t}\\n\\tdefer file.Close()\\n\\n\\t// Seek to end of file to only process new entries\\n\\tif _, err := file.Seek(0, 2); err != nil {\\n\\t\\treturn fmt.Errorf(\"failed to seek to end of log file: %w\", err)\\n\\t}\\n\\n\\treader := bufio.NewReader(file)\\n\\tfor {\\n\\t\\tselect {\\n\\t\\tcase <-ctx.Done():\\n\\t\\t\\tlog.Println(\"Context cancelled, stopping detector\")\\n\\t\\t\\treturn nil\\n\\t\\tdefault:\\n\\t\\t\\tline, err := reader.ReadString('\\\\n')\\n\\t\\t\\tif err != nil {\\n\\t\\t\\t\\tif err.Error() == \"EOF\" {\\n\\t\\t\\t\\t\\ttime.Sleep(100 * time.Millisecond)\\n\\t\\t\\t\\t\\tcontinue\\n\\t\\t\\t\\t}\\n\\t\\t\\t\\treturn fmt.Errorf(\"error reading log line: %w\", err)\\n\\t\\t\\t}\\n\\t\\t\\td.processLogLine(line)\\n\\t\\t}\\n\\t}\\n}\\n\\n// processLogLine checks if a log line contains a deadlock and applies mitigation\\nfunc (d *DeadlockDetector) processLogLine(line string) {\\n\\tif !d.deadlock.MatchString(line) {\\n\\t\\treturn\\n\\t}\\n\\tlog.Println(\"Detected deadlock in log line:\", strings.TrimSpace(line))\\n\\n\\t// Extract the PID of the deadlocked process (simplified extraction)\\n\\tpidRegex := regexp.MustCompile(`process (\\\\d+)`)\\n\\tmatches := pidRegex.FindStringSubmatch(line)\\n\\tif len(matches) < 2 {\\n\\t\\tlog.Println(\"Failed to extract PID from deadlock line\")\\n\\t\\treturn\\n\\t}\\n\\tpid := matches[1]\\n\\tif d.mitigated[pid] {\\n\\t\\tlog.Printf(\"PID %s already mitigated, skipping\", pid)\\n\\t\\treturn\\n\\t}\\n\\n\\t// Apply mitigation: terminate the deadlocked PID and disable parallel scans for the user\\n\\tif err := d.mitigatePID(pid); err != nil {\\n\\t\\tlog.Printf(\"Failed to mitigate PID %s: %v\", pid, err)\\n\\t\\treturn\\n\\t}\\n\\td.mitigated[pid] = true\\n\\tlog.Printf(\"Successfully mitigated deadlock for PID %s\", pid)\\n}\\n\\n// mitigatePID terminates the deadlocked process and applies session-level fixes\\nfunc (d *DeadlockDetector) mitigatePID(pid string) error {\\n\\ttx, err := d.db.Begin()\\n\\tif err != nil {\\n\\t\\treturn fmt.Errorf(\"failed to begin transaction: %w\", err)\\n\\t}\\n\\tdefer tx.Rollback()\\n\\n\\t// First, terminate the deadlocked process\\n\\t_, err = tx.Exec(fmt.Sprintf(\"SELECT pg_terminate_backend(%s)\", pid))\\n\\tif err != nil {\\n\\t\\treturn fmt.Errorf(\"failed to terminate PID %s: %w\", pid, err)\\n\\t}\\n\\n\\t// Get the user associated with the deadlocked PID\\n\\tvar username string\\n\\terr = tx.QueryRow(fmt.Sprintf(`\\n\\t\\tSELECT usename FROM pg_stat_activity WHERE pid = %s\\n\\t`, pid)).Scan(&username)\\n\\tif err != nil {\\n\\t\\tlog.Printf(\"Failed to get username for PID %s: %v\", pid, err)\\n\\t\\t// Still return nil because we terminated the PID successfully\\n\\t\\treturn nil\\n\\t}\\n\\n\\t// Disable parallel sequential scans for the user's future sessions\\n\\t_, err = tx.Exec(fmt.Sprintf(`\\n\\t\\tALTER USER %s SET max_parallel_workers_per_gather = 0\\n\\t`, username))\\n\\tif err != nil {\\n\\t\\treturn fmt.Errorf(\"failed to alter user %s: %w\", username, err)\\n\\t}\\n\\n\\treturn tx.Commit()\\n}\\n\\nfunc main() {\\n\\t// Configuration - update these values for your environment\\n\\tcfg := DBConfig{\\n\\t\\tHost:     \"localhost\",\\n\\t\\tPort:     5432,\\n\\t\\tUser:     \"postgres\",\\n\\t\\tPassword: \"postgres\",\\n\\t\\tDBName:   \"deadlock_test\",\\n\\t}\\n\\tlogPath := \"/var/log/postgresql/postgresql-16-main.log\"\\n\\n\\tdetector, err := NewDeadlockDetector(cfg, logPath)\\n\\tif err != nil {\\n\\t\\tlog.Fatalf(\"Failed to initialize detector: %v\", err)\\n\\t}\\n\\n\\t// Handle shutdown signals\\n\\tctx, cancel := context.WithCancel(context.Background())\\n\\tsigChan := make(chan os.Signal, 1)\\n\\tsignal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)\\n\\tgo func() {\\n\\t\\t<-sigChan\\n\\t\\tlog.Println(\"Received shutdown signal\")\\n\\t\\tcancel()\\n\\t}()\\n\\n\\tlog.Println(\"Starting deadlock detector...\")\\n\\tif err := detector.Start(ctx); err != nil {\\n\\t\\tlog.Fatalf(\"Detector failed: %v\", err)\\n\\t}\\n}\\n

\\n#!/bin/bash\\n\\n# PostgreSQL 15 vs 16 Deadlock Benchmark Script\\n# Requires: pgbench, PostgreSQL 15 and 16 instances running on ports 5432 and 5433 respectively\\n# Usage: ./benchmark.sh\\n\\nset -euo pipefail\\n\\n# Configuration\\nPG15_PORT=5432\\nPG16_PORT=5433\\nDB_NAME=\"deadlock_bench\"\\nUSER=\"postgres\"\\nPASSWORD=\"postgres\"\\nPG15_CONN=\"host=localhost port=$PG15_PORT dbname=$DB_NAME user=$USER password=$PASSWORD sslmode=disable\"\\nPG16_CONN=\"host=localhost port=$PG16_PORT dbname=$DB_NAME user=$USER password=$PASSWORD sslmode=disable\"\\nBENCH_DURATION=300  # 5 minutes per benchmark run\\nCLIENTS=(4 8 16 32) # Number of concurrent clients\\nTHREADS=4\\nRESULTS_DIR=\"./benchmark_results\"\\nmkdir -p \"$RESULTS_DIR\"\\n\\n# Log function\\nlog() {\\n    echo \"[$(date +'%Y-%m-%dT%H:%M:%S%z')] $*\"\\n}\\n\\n# Initialize test database for a given port\\ninitialize_db() {\\n    local port=$1\\n    local conn=\"host=localhost port=$port dbname=postgres user=$USER password=$PASSWORD sslmode=disable\"\\n    log \"Initializing database $DB_NAME on port $port...\"\\n    \\n    # Create test database\\n    psql \"$conn\" -c \"DROP DATABASE IF EXISTS $DB_NAME\"\\n    psql \"$conn\" -c \"CREATE DATABASE $DB_NAME\"\\n    \\n    # Create test table and seed data\\n    psql \"$conn\" -d \"$DB_NAME\" < \"$script_file\" < \"$output_file\" 2>&1\\n\\n    # Count deadlocks in pgbench output\\n    local deadlocks=$(grep -i \"deadlock detected\" \"$output_file\" | wc -l)\\n    echo \"$deadlocks\" > \"${output_file}.deadlocks\"\\n    log \"Benchmark on port $port with $clients clients completed. Deadlocks: $deadlocks\"\\n}\\n\\n# Main execution\\nlog \"Starting benchmark run...\"\\n\\n# Initialize databases\\ninitialize_db \"$PG15_PORT\"\\ninitialize_db \"$PG16_PORT\"\\n\\n# Run benchmarks for each client count on both versions\\nfor clients in \"${CLIENTS[@]}\"; do\\n    # PostgreSQL 15 benchmark\\n    run_bench \"$PG15_PORT\" \"$clients\" \"$RESULTS_DIR/pg15_${clients}_clients.txt\"\\n    \\n    # PostgreSQL 16 benchmark (before fix)\\n    run_bench \"$PG16_PORT\" \"$clients\" \"$RESULTS_DIR/pg16_before_${clients}_clients.txt\"\\n    \\n    # Apply fix to PG16 (disable parallel workers)\\n    log \"Applying fix to PG16 (disable parallel sequential scans)...\"\\n    psql \"$PG16_CONN\" -c \"ALTER SYSTEM SET max_parallel_workers_per_gather = 0;\"\\n    psql \"$PG16_CONN\" -c \"SELECT pg_reload_conf();\"\\n    \\n    # PostgreSQL 16 benchmark (after fix)\\n    run_bench \"$PG16_PORT\" \"$clients\" \"$RESULTS_DIR/pg16_after_${clients}_clients.txt\"\\n    \\n    # Revert PG16 setting for next run\\n    psql \"$PG16_CONN\" -c \"ALTER SYSTEM SET max_parallel_workers_per_gather = 2;\"\\n    psql \"$PG16_CONN\" -c \"SELECT pg_reload_conf();\"\\ndone\\n\\n# Generate summary report\\nlog \"Generating summary report...\"\\nsummary_file=\"$RESULTS_DIR/summary.txt\"\\necho \"PostgreSQL 15 vs 16 Deadlock Benchmark Results\" > \"$summary_file\"\\necho \"==============================================\" >> \"$summary_file\"\\necho \"Benchmark duration: $BENCH_DURATION seconds per run\" >> \"$summary_file\"\\necho \"\" >> \"$summary_file\"\\n\\nfor clients in \"${CLIENTS[@]}\"; do\\n    pg15_deadlocks=$(cat \"$RESULTS_DIR/pg15_${clients}_clients.txt.deadlocks\")\\n    pg16_before_deadlocks=$(cat \"$RESULTS_DIR/pg16_before_${clients}_clients.txt.deadlocks\")\\n    pg16_after_deadlocks=$(cat \"$RESULTS_DIR/pg16_after_${clients}_clients.txt.deadlocks\")\\n    \\n    echo \"Clients: $clients\" >> \"$summary_file\"\\n    echo \"  PG15 Deadlocks: $pg15_deadlocks\" >> \"$summary_file\"\\n    echo \"  PG16 Before Fix Deadlocks: $pg16_before_deadlocks\" >> \"$summary_file\"\\n    echo \"  PG16 After Fix Deadlocks: $pg16_after_deadlocks\" >> \"$summary_file\"\\n    echo \"\" >> \"$summary_file\"\\ndone\\n\\nlog \"Benchmark complete. Results saved to $RESULTS_DIR\"\\ncat \"$summary_file\"\\n

-- Check for active deadlocks or lock waits in PostgreSQL\\nSELECT \\n    pid, \\n    usename, \\n    query, \\n    wait_event_type, \\n    wait_event, \\n    now() - query_start AS duration\\nFROM pg_stat_activity \\nWHERE wait_event_type = 'Lock' \\n  AND state = 'active';\\n

-- Apply session-level fix for a single OLAP query\\nBEGIN;\\n-- Disable parallel sequential scans for this transaction only\\nSET LOCAL max_parallel_workers_per_gather = 0;\\n-- Run the OLAP query that previously triggered deadlocks\\nSELECT status, COUNT(*), SUM(amount) FROM transaction_logs GROUP BY status;\\nCOMMIT;\\n\\n-- Permanently disable parallel scans for the reporting service user\\nALTER USER reporting_service SET max_parallel_workers_per_gather = 0;\\n

-- Run pgBadger to generate a deadlock report from PostgreSQL logs\\npgbadger -f stderr --prefix '%t [%p]: [%l-1] user=%u,db=%d,app=%a,client=%h ' \\\\\\n  /var/log/postgresql/postgresql-16-main.log \\\\\\n  -o /var/www/html/pgbadger_report.html\\n\\n-- The report will include a dedicated \"Deadlocks\" section with frequency, queries, and PIDs\\n