transactions

name	transactions
description	Monitor database transactions with real-time alerting for performance and...
shortcut	txnm

Database Transaction Monitor

Monitor database transaction performance, detect long-running transactions, identify lock contention, track rollback rates, and automatically alert on transaction anomalies for production database health.

When to Use This Command

Use /txn-monitor when you need to:

• Detect and kill long-running transactions blocking other queries
• Monitor lock wait times and identify deadlock patterns
• Track transaction rollback rates for error analysis
• Alert on isolation level anomalies (phantom reads, dirty reads)
• Analyze transaction throughput and latency trends
• Investigate application connection leak issues

DON'T use this when:

• Database has minimal transaction load (<100 TPS)
• All transactions complete within milliseconds
• Looking for query optimization (use query optimizer instead)
• Investigating data corruption (use audit logger instead)

Design Decisions

This command implements real-time transaction monitoring with automated alerting because:

• Long-running transactions (>30s) block other queries and cause performance degradation
• Lock contention detection prevents cascade failures
• Rollback rate monitoring identifies application bugs early
• Automatic alerts reduce MTTR (Mean Time To Resolution)
• Historical trend analysis enables capacity planning

Alternative considered: Periodic manual checks

• No automated alerting on issues
• Relies on humans checking dashboards
• Slower incident response
• Recommended only for development environments

Alternative considered: Database log parsing

• Post-mortem analysis only
• No real-time alerts
• Requires custom log parsing logic
• Recommended for compliance/audit purposes

Prerequisites

Before running this command:

1. Database monitoring permissions (pg_monitor role or PROCESS privilege)
2. Access to pg_stat_activity (PostgreSQL) or performance_schema (MySQL)
3. Alerting infrastructure (Slack, PagerDuty, email)
4. Monitoring data retention strategy (metrics database or time-series DB)
5. Runbook for common transaction issues

Implementation Process

Step 1: Enable Transaction Monitoring

Configure database to track transaction statistics.

Step 2: Build Real-Time Monitor

Create monitoring script that polls transaction statistics every 5-10 seconds.

Step 3: Define Alert Thresholds

Set thresholds for long-running transactions, lock waits, and rollback rates.

Step 4: Implement Automated Actions

Auto-kill transactions exceeding thresholds or alert operators.

Step 5: Create Dashboards

Build Grafana dashboards for transaction metrics visualization.

Output Format

The command generates:

• monitoring/transaction_monitor.py - Real-time transaction monitoring daemon
• queries/transaction_analysis.sql - Transaction health diagnostic queries
• alerts/transaction_alerts.yml - Prometheus alerting rules
• dashboards/transaction_dashboard.json - Grafana dashboard configuration
• docs/transaction_runbook.md - Incident response procedures

Code Examples

Example 1: PostgreSQL Real-Time Transaction Monitor

# monitoring/postgres_transaction_monitor.py
import psycopg2
from psycopg2.extras import Dict Cursor
import time
import logging
from typing import List, Dict, Optional
from dataclasses import dataclass, asdict
from datetime import datetime, timedelta

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class TransactionInfo:
    """Represents an active transaction."""
    pid: int
    username: str
    database: str
    application_name: str
    client_addr: str
    state: str
    query: str
    transaction_start: datetime
    query_start: datetime
    wait_event: Optional[str]
    blocking_pids: List[int]

    def duration_seconds(self) -> float:
        return (datetime.now() - self.transaction_start).total_seconds()

    def to_dict(self) -> dict:
        result = asdict(self)
        result['transaction_start'] = self.transaction_start.isoformat()
        result['query_start'] = self.query_start.isoformat()
        result[] = .duration_seconds()
         result

 :
    

     ():
        .conn_string = connection_string
        .long_transaction_threshold = long_transaction_threshold
        .check_interval = check_interval
        .stats = {
            : ,
            : ,
            : ,
            : 
        }

     ():
         psycopg2.connect(.conn_string, cursor_factory=DictCursor)

     () -> [TransactionInfo]:
        
        query = 

        conn = .connect()
        :
             conn.cursor()  cur:
                cur.execute(query)
                rows = cur.fetchall()

                transactions = []
                 row  rows:
                    txn = TransactionInfo(
                        pid=row[],
                        username=row[],
                        database=row[],
                        application_name=row[]  ,
                        client_addr=row[]  ,
                        state=row[],
                        query=row[][:],  
                        transaction_start=row[],
                        query_start=row[],
                        wait_event=row[],
                        blocking_pids=row[]  []
                    )
                    transactions.append(txn)

                 transactions

        :
            conn.close()

     () -> [TransactionInfo]:
        
         [
            txn  txn  transactions
             txn.duration_seconds() > .long_transaction_threshold
        ]

     () -> [TransactionInfo]:
        
         [
            txn  txn  transactions
             txn.blocking_pids  (txn.blocking_pids) > 
        ]

     () -> [TransactionInfo]:
        
         [
            txn  txn  transactions
             txn.state == 
             txn.duration_seconds() > 
        ]

     () -> :
        
        conn = .connect()
        :
             conn.cursor()  cur:
                cur.execute(, (pid,))
                success = cur.fetchone()[]

                 success:
                    logger.warning()
                :
                    logger.error()

                 success

        :
            conn.close()

     () -> [, ]:
        
        conn = .connect()
        :
             conn.cursor()  cur:
                cur.execute()

                row = cur.fetchone()

                total_txns = row[] + row[]
                rollback_rate = (row[] / total_txns * )  total_txns >   

                 {
                    : row[],
                    : row[],
                    : row[],
                    : row[],
                    : (rollback_rate, ),
                    : row[]
                }

        :
            conn.close()

     ():
        
        log_func = {
            : logger.critical,
            : logger.warning,
            : logger.info
        }.get(severity, logger.info)

        log_func()

         details:
            logger.info()

        
        

     ():
        
        logger.info()

         :
            :
                
                transactions = .get_active_transactions()
                .stats[] = (transactions)

                
                long_running = .find_long_running_transactions(transactions)
                 long_running:
                    .stats[] = (long_running)

                     txn  long_running:
                        .alert(
                            ,
                            ,
                            {
                                : txn.duration_seconds(),
                                : txn.database,
                                : txn.username,
                                : txn.query
                            }
                        )

                        
                         txn.duration_seconds() > :
                            .kill_transaction(
                                txn.pid,
                                
                            )

                
                blocked = .find_blocked_transactions(transactions)
                 blocked:
                    .stats[] = (blocked)

                     txn  blocked:
                        .alert(
                            ,
                            ,
                            {
                                : txn.blocking_pids,
                                : txn.wait_event,
                                : txn.duration_seconds()
                            }
                        )

                
                idle_txns = .find_idle_in_transaction(transactions)
                 idle_txns:
                    .stats[] = (idle_txns)

                     txn  idle_txns:
                        .alert(
                            ,
                            ,
                            {
                                : txn.duration_seconds(),
                                : txn.application_name
                            }
                        )

                        
                         txn.duration_seconds() > :
                            .kill_transaction(txn.pid, )

                
                stats = .get_transaction_stats()

                
                 stats[] > :
                    .alert(
                        ,
                        ,
                        stats
                    )

                
                logger.info(
                    
                    
                    
                    
                )

                time.sleep(.check_interval)

             KeyboardInterrupt:
                logger.info()
                
             Exception  e:
                logger.error()
                time.sleep(.check_interval)


 __name__ == :
    monitor = PostgreSQLTransactionMonitor(
        connection_string=,
        long_transaction_threshold=,  
        check_interval=  
    )

    monitor.run_monitoring_loop()

Example 2: Transaction Analysis Queries

-- PostgreSQL transaction health diagnostic queries

-- 1. Long-running transactions
SELECT
    pid,
    usename,
    application_name,
    client_addr,
    NOW() - xact_start AS transaction_duration,
    NOW() - query_start AS query_duration,
    state,
    LEFT(query, 100) AS query_snippet
FROM pg_stat_activity
WHERE xact_start IS NOT NULL
  AND state != 'idle'
  AND pid != pg_backend_pid()
ORDER BY xact_start;

-- 2. Blocking tree (which transactions are blocking others)
WITH RECURSIVE blocking_tree AS (
    SELECT
        a.pid,
        a.usename,
        a.query AS blocked_query,
        NULL::integer AS blocking_pid,
        NULL::text AS blocking_query,
        1 AS level
    FROM pg_stat_activity a
    WHERE NOT EXISTS (
        SELECT 1 FROM pg_stat_activity b
        WHERE b.pid = ANY(pg_blocking_pids(a.pid))
    )
    AND a.pid IN (
        SELECT unnest(pg_blocking_pids(c.pid))
         pg_stat_activity c
    )

     

    
        a.pid,
        a.usename,
        a.query,
        b.pid,
        b.query,
        bt.level  
     blocking_tree bt
     pg_stat_activity a  a.pid  (
         (pg_blocking_pids(x.pid))
         pg_stat_activity x
         x.pid  bt.pid
    )
     pg_stat_activity b  b.pid  (pg_blocking_pids(a.pid))
)

    level,
    pid,
    usename,
    blocking_pid,
    (blocked_query, )  blocked_query,
    (blocking_query, )  blocking_query
 blocking_tree
  level, pid;



    datname,
    xact_commit  commits,
    xact_rollback  rollbacks,
    ROUND(  xact_rollback  (xact_commit  xact_rollback, ), )  rollback_rate_percent
 pg_stat_database
 datname   (, , )
  rollback_rate_percent ;



    pid,
    usename,
    application_name,
    client_addr,
    NOW()  state_change  idle_duration,
    state,
    query
 pg_stat_activity
 state  
   pid  pg_backend_pid()
  state_change;



    wait_event_type,
    wait_event,
    ()  waiting_count,
    ( pid)  waiting_pids
 pg_stat_activity
 wait_event  
   state  
  wait_event_type, wait_event
  waiting_count ;

Error Handling

Error	Cause	Solution
"Permission denied for pg_stat_activity"	Insufficient monitoring privileges	Grant pg_monitor role or SELECT on pg_stat_activity
"Cannot terminate backend"	Trying to kill superuser connection	Use pg_cancel_backend or kill from OS level
"Connection pool exhausted"	Too many idle connections	Kill idle in transaction connections, increase pool size
"High rollback rate"	Application errors or constraint violations	Review application logs and fix bugs
"Lock wait timeout exceeded"	Deadlock or very long lock hold	Analyze blocking queries, implement timeouts

Configuration Options

Monitoring Intervals

• check_interval: 5-10 seconds for real-time alerting
• long_transaction_threshold: 30-60 seconds (production), 300s (analytics)
• idle_in_transaction_timeout: 600 seconds (10 minutes)

Auto-Kill Thresholds

• Long-running OLTP: 60-300 seconds
• Long-running analytics: 3600 seconds (1 hour)
• Idle in transaction: 600 seconds (10 minutes)

Alert Thresholds

• Rollback rate: >5% warning, >10% critical
• Blocked transactions: >10 warning, >50 critical
• Active connections: >80% of max_connections

Best Practices

DO:

• Set statement_timeout in application connection strings
• Use connection pooling to limit total connections
• Implement transaction timeout in application code
• Monitor transaction throughput trends over time
• Kill idle in transaction connections automatically
• Track rollback reasons in application logs

DON'T:

• Leave transactions open while waiting for user input
• Hold locks during expensive operations (file I/O, network calls)
• Use long-running transactions in OLTP workloads
• Ignore idle in transaction connections (they hold locks)
• Set transaction timeouts too low (causes false positives)

Performance Considerations

• Monitoring adds <0.1% CPU overhead with 10-second intervals
• pg_stat_activity queries are lightweight (<1ms)
• Auto-killing transactions requires careful threshold tuning
• Historical metrics retention: 30 days (aggregated), 7 days (detailed)
• Consider read replicas for monitoring queries in high-load systems

Related Commands

• /database-deadlock-detector - Detailed deadlock analysis
• /database-health-monitor - Overall database health metrics
• /sql-query-optimizer - Optimize slow queries causing lock contention
• /database-connection-pooler - Manage connection pool sizing

Version History

• v1.0.0 (2024-10): Initial implementation with PostgreSQL real-time monitoring
• Planned v1.1.0: Add MySQL transaction monitoring and distributed transaction support