# CockroachDB Transaction Handling Rule ## Overview This rule provides comprehensive guidance for handling transactions in CockroachDB, focusing on serializable isolation, retry logic, and optimistic concurrency control. CockroachDB uses serializable isolation by default, which requires specific patterns to handle transaction conflicts and retries effectively. ## Implementation ### 1. Basic Transaction Patterns #### Simple Transaction Block ```sql -- Basic transaction syntax BEGIN; INSERT INTO accounts (id, balance) VALUES (1, 1000); INSERT INTO accounts (id, balance) VALUES (2, 500); COMMIT; ``` #### Transaction with Savepoints ```sql BEGIN; SAVEPOINT before_transfer; UPDATE accounts SET balance = balance - 100 WHERE id = 1; UPDATE accounts SET balance = balance + 100 WHERE id = 2; -- Conditional rollback IF (SELECT balance FROM accounts WHERE id = 1) < 0 THEN ROLLBACK TO SAVEPOINT before_transfer; END IF; COMMIT; ``` ### 2. Application-Level Retry Logic #### Go Implementation with Retry ```go package main import ( "context" "database/sql" "fmt" "time" "github.com/cockroachdb/cockroach-go/v2/crdb/crdbpgx" "github.com/jackc/pgx/v5/pgxpool" ) func transferMoney(ctx context.Context, db *pgxpool.Pool, fromAccount, toAccount int, amount int) error { return crdbpgx.ExecuteTx(ctx, db, pgx.TxOptions{}, func(tx pgx.Tx) error { // Check source account balance var balance int err := tx.QueryRow(ctx, "SELECT balance FROM accounts WHERE id = $1", fromAccount).Scan(&balance) if err != nil { return fmt.Errorf("failed to get balance: %w", err) } if balance < amount { return fmt.Errorf("insufficient funds: have %d, need %d", balance, amount) } // Perform the transfer _, err = tx.Exec(ctx, "UPDATE accounts SET balance = balance - $1 WHERE id = $2", amount, fromAccount) if err != nil { return fmt.Errorf("failed to debit account: %w", err) } _, err = tx.Exec(ctx, "UPDATE accounts SET balance = balance + $1 WHERE id = $2", amount, toAccount) if err != nil { return fmt.Errorf("failed to credit account: %w", err) } return nil }) } // Manual retry implementation func transferWithManualRetry(ctx context.Context, db *pgxpool.Pool, fromAccount, toAccount int, amount int) error { maxRetries := 5 baseDelay := 100 * time.Millisecond for i := 0; i < maxRetries; i++ { err := transferMoney(ctx, db, fromAccount, toAccount, amount) if err == nil { return nil } // Check if it's a retryable error if !isRetryableError(err) { return err } // Exponential backoff with jitter delay := time.Duration(float64(baseDelay) * math.Pow(2, float64(i))) jitter := time.Duration(rand.Float64() * float64(delay) * 0.1) time.Sleep(delay + jitter) } return fmt.Errorf("transaction failed after %d retries", maxRetries) } func isRetryableError(err error) bool { // Check for serialization errors return strings.Contains(err.Error(), "restart transaction") || strings.Contains(err.Error(), "RETRY_WRITE_TOO_OLD") || strings.Contains(err.Error(), "RETRY_SERIALIZABLE") } ``` #### Python Implementation with Retry ```python import psycopg2 import time import random from contextlib import contextmanager class RetryableTransactionError(Exception): pass @contextmanager def get_db_connection(): conn = psycopg2.connect( host="localhost", port=26257, database="myapp", user="app_user", password="password", sslmode="require" ) try: yield conn finally: conn.close() def transfer_money_with_retry(from_account, to_account, amount, max_retries=5): for attempt in range(max_retries): try: with get_db_connection() as conn: with conn.cursor() as cur: # Start transaction cur.execute("BEGIN") # Check balance cur.execute("SELECT balance FROM accounts WHERE id = %s", (from_account,)) balance = cur.fetchone()[0] if balance < amount: raise ValueError(f"Insufficient funds: {balance} < {amount}") # Perform transfer cur.execute( "UPDATE accounts SET balance = balance - %s WHERE id = %s", (amount, from_account) ) cur.execute( "UPDATE accounts SET balance = balance + %s WHERE id = %s", (amount, to_account) ) # Commit transaction cur.execute("COMMIT") return True except psycopg2.Error as e: if is_retryable_error(e): if attempt < max_retries - 1: # Exponential backoff with jitter delay = (2 ** attempt) * 0.1 + random.uniform(0, 0.1) time.sleep(delay) continue raise e raise RetryableTransactionError("Transaction failed after all retries") def is_retryable_error(error): error_msg = str(error) return any(msg in error_msg for msg in [ "restart transaction", "RETRY_WRITE_TOO_OLD", "RETRY_SERIALIZABLE" ]) ``` ### 3. Advanced Transaction Patterns #### Batch Processing with Transactions ```sql -- Process records in batches to avoid long-running transactions DO $$ DECLARE batch_size INTEGER := 1000; processed INTEGER := 0; total_rows INTEGER; BEGIN SELECT COUNT(*) INTO total_rows FROM pending_orders; WHILE processed < total_rows LOOP BEGIN -- Process batch WITH batch AS ( SELECT id FROM pending_orders ORDER BY id LIMIT batch_size OFFSET processed ) UPDATE orders SET status = 'processed' WHERE id IN (SELECT id FROM batch); processed := processed + batch_size; -- Log progress RAISE NOTICE 'Processed % of % records', processed, total_rows; EXCEPTION WHEN serialization_failure THEN -- Retry the current batch RAISE NOTICE 'Retrying batch starting at %', processed; END; END LOOP; END $$; ``` #### Conditional Transactions ```sql -- Transaction with conditional logic BEGIN; -- Get current inventory SELECT quantity INTO @current_qty FROM inventory WHERE product_id = 123; -- Only proceed if we have enough stock IF @current_qty >= 5 THEN UPDATE inventory SET quantity = quantity - 5 WHERE product_id = 123; INSERT INTO orders (product_id, quantity, status) VALUES (123, 5, 'confirmed'); ELSE -- Log insufficient stock INSERT INTO audit_log (event, product_id, message) VALUES ('insufficient_stock', 123, 'Attempted to order 5, only ' || @current_qty || ' available'); ROLLBACK; END IF; COMMIT; ``` ### 4. Performance Optimization #### Use SELECT FOR UPDATE Wisely ```sql -- Lock specific rows to prevent conflicts BEGIN; SELECT balance FROM accounts WHERE id = 1 FOR UPDATE; SELECT balance FROM accounts WHERE id = 2 FOR UPDATE; -- Perform updates with locked rows UPDATE accounts SET balance = balance - 100 WHERE id = 1; UPDATE accounts SET balance = balance + 100 WHERE id = 2; COMMIT; ``` #### Minimize Transaction Scope ```go // Bad: Long-running transaction func badExample(ctx context.Context, db *pgxpool.Pool) error { tx, err := db.Begin(ctx) if err != nil { return err } defer tx.Rollback(ctx) // Long computation inside transaction result := performComplexCalculation() // External API call inside transaction apiResponse, err := callExternalAPI() if err != nil { return err } // Database operation _, err = tx.Exec(ctx, "UPDATE table SET value = $1", result) if err != nil { return err } return tx.Commit(ctx) } // Good: Minimal transaction scope func goodExample(ctx context.Context, db *pgxpool.Pool) error { // Perform computations outside transaction result := performComplexCalculation() apiResponse, err := callExternalAPI() if err != nil { return err } // Minimal transaction scope return crdbpgx.ExecuteTx(ctx, db, pgx.TxOptions{}, func(tx pgx.Tx) error { _, err := tx.Exec(ctx, "UPDATE table SET value = $1", result) return err }) } ``` ### 5. Transaction Isolation Levels #### Working with Different Isolation Levels ```sql -- Default serializable isolation BEGIN ISOLATION LEVEL SERIALIZABLE; -- All operations are serializable SELECT * FROM accounts WHERE id = 1; UPDATE accounts SET balance = balance - 100 WHERE id = 1; COMMIT; -- Read committed for read-heavy workloads BEGIN ISOLATION LEVEL READ COMMITTED; -- Reduced contention for read operations SELECT * FROM reporting_view WHERE date >= '2024-01-01'; COMMIT; ``` ## Best Practices ### 1. Transaction Design - **Keep transactions short**: Minimize the time between BEGIN and COMMIT - **Batch operations**: Group related operations in single transactions - **Avoid external calls**: Never make HTTP requests or file I/O inside transactions - **Use appropriate isolation**: Choose READ COMMITTED for read-heavy operations ### 2. Retry Strategy - **Implement exponential backoff**: Use increasing delays between retries - **Add jitter**: Randomize retry delays to prevent thundering herd - **Limit retry attempts**: Set maximum retry counts to prevent infinite loops - **Monitor retry rates**: Track retry frequency to identify contention issues ### 3. Error Handling - **Distinguish error types**: Handle retryable vs non-retryable errors differently - **Log retry attempts**: Track retry patterns for debugging - **Implement circuit breakers**: Stop retrying after repeated failures - **Use proper exception types**: Create specific exception classes for different error scenarios ### 4. Performance Considerations - **Order operations consistently**: Always access tables in the same order to prevent deadlocks - **Use SELECT FOR UPDATE sparingly**: Only when necessary to prevent conflicts - **Minimize lock scope**: Lock only the rows you need to modify - **Consider async processing**: Use queues for operations that don't need immediate consistency ## Common Issues ### 1. Serialization Failures **Problem**: High frequency of "restart transaction" errors **Solution**: ```sql -- Check transaction retry rates SELECT application_name, COUNT(*) as retry_count, AVG(service_lat) as avg_latency FROM crdb_internal.node_statement_statistics WHERE key LIKE '%restart transaction%' GROUP BY application_name ORDER BY retry_count DESC; ``` ### 2. Deadlock Detection **Problem**: Transactions waiting indefinitely **Solution**: ```sql -- Monitor long-running transactions SELECT query, start, NOW() - start as duration, state FROM crdb_internal.cluster_queries WHERE NOW() - start > INTERVAL '30 seconds' ORDER BY duration DESC; ``` ### 3. Transaction Timeout Issues **Problem**: Transactions failing due to timeouts **Solution**: ```go // Set appropriate context timeout ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() // Use shorter timeout for quick operations quickCtx, quickCancel := context.WithTimeout(context.Background(), 5*time.Second) defer quickCancel() ``` ### 4. Memory Issues with Large Transactions **Problem**: Out of memory errors during large batch operations **Solution**: ```sql -- Process in smaller batches DO $$ DECLARE batch_size INTEGER := 1000; -- Reduced batch size offset_val INTEGER := 0; row_count INTEGER; BEGIN LOOP -- Process batch UPDATE large_table SET processed = true WHERE id IN ( SELECT id FROM large_table WHERE processed = false ORDER BY id LIMIT batch_size OFFSET offset_val ); GET DIAGNOSTICS row_count = ROW_COUNT; EXIT WHEN row_count = 0; offset_val := offset_val + batch_size; -- Allow other transactions to proceed PERFORM pg_sleep(0.1); END LOOP; END $$; ``` ### 5. Connection Pool Exhaustion **Problem**: Connection pool exhausted during high retry rates **Solution**: ```go // Implement connection pool monitoring func monitorConnectionPool(pool *pgxpool.Pool) { go func() { ticker := time.NewTicker(30 * time.Second) defer ticker.Stop() for range ticker.C { stats := pool.Stat() log.Printf("Pool stats: Total=%d, Idle=%d, Used=%d", stats.TotalConns(), stats.IdleConns(), stats.AcquiredConns()) if float64(stats.AcquiredConns())/float64(stats.TotalConns()) > 0.8 { log.Warn("Connection pool utilization high") } } }() } ``` ## Monitoring and Debugging ### Transaction Performance Queries ```sql -- Monitor transaction statistics SELECT application_name, statement_type, count, mean_lat, p99_lat, max_retries FROM crdb_internal.statement_statistics WHERE statement_type = 'TypeDML' ORDER BY mean_lat DESC; -- Check for transaction conflicts SELECT waiting_query, blocking_query, wait_start, NOW() - wait_start as wait_duration FROM crdb_internal.cluster_locks WHERE NOW() - wait_start > INTERVAL '10 seconds'; ```