How to fix SQLITE_BUSY_SNAPSHOT: Cannot promote read transaction to write transaction in SQLite

-- DEFERRED (default): Transaction starts as read, can upgrade to write
BEGIN DEFERRED;

-- IMMEDIATE: Transaction starts as write, prevents other writes
BEGIN IMMEDIATE;

-- EXCLUSIVE: Exclusive lock on database
BEGIN EXCLUSIVE;

-- SNAPSHOT: Read-only snapshot, cannot upgrade to write
BEGIN SNAPSHOT;

BEGIN SNAPSHOT;
SELECT * FROM users WHERE id = 1;
-- Later in code...
UPDATE users SET name = 'John' WHERE id = 1; -- ERROR!
COMMIT;

BEGIN DEFERRED;
SELECT * FROM users WHERE id = 1;
-- Can now write...
UPDATE users SET name = 'John' WHERE id = 1; -- SUCCESS
COMMIT;

import sqlite3

# Wrong - using SNAPSHOT when writes might occur
conn = sqlite3.connect('database.db')
conn.execute('BEGIN SNAPSHOT')  # Will fail if writing later

# Correct - use DEFERRED (default) when writes might occur
conn = sqlite3.connect('database.db')
conn.execute('BEGIN')  # Defaults to DEFERRED, can upgrade to write

const Database = require('better-sqlite3');
const db = new Database('database.db');

// Wrong
const transaction = db.transaction(() => {
  // This uses SNAPSHOT internally if unspecified
});

// Correct - specify DEFERRED
const transaction = db.transaction(() => {
  // Your code
});
transaction.immediate = false; // Use DEFERRED mode

def update_user(user_id, new_name):
    # 1. Read with SNAPSHOT
    with db_connection() as conn:
        conn.execute('BEGIN SNAPSHOT')
        cursor = conn.execute('SELECT * FROM users WHERE id = ?', (user_id,))
        user = cursor.fetchone()
        conn.execute('COMMIT')

    # 2. Write with DEFERRED
    with db_connection() as conn:
        conn.execute('BEGIN DEFERRED')
        conn.execute('UPDATE users SET name = ? WHERE id = ?', (new_name, user_id))
        conn.execute('COMMIT')

-- If you know you'll write, start with IMMEDIATE
BEGIN IMMEDIATE;
UPDATE users SET last_login = CURRENT_TIMESTAMP WHERE id = 1;
COMMIT;

// Bad: Long transaction that might need to write
db.transaction(() => {
  const user = db.prepare('SELECT * FROM users WHERE id = ?').get(userId);
  // ... lots of other code ...
  if (user.needsUpdate) {
    db.prepare('UPDATE users SET updated = 1 WHERE id = ?').run(userId); // ERROR!
  }
});

// Good: Separate transactions
const user = db.prepare('SELECT * FROM users WHERE id = ?').get(userId);
// ... logic ...
if (user.needsUpdate) {
  db.transaction(() => {
    db.prepare('UPDATE users SET updated = 1 WHERE id = ?').run(userId);
  });
}

PRAGMA busy_timeout = 30000; -- Wait up to 30 seconds for locks

PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;

import sqlite3

def get_write_connection(db_path):
    conn = sqlite3.connect(db_path)
    # Start transaction immediately to avoid promotion issues
    conn.execute('BEGIN IMMEDIATE')
    return conn

def release_write_connection(conn):
    conn.commit()
    conn.close()

class ConnectionPool {
  constructor(dbPath) {
    this.readConnections = []; // For SNAPSHOT/read-only
    this.writeConnections = []; // For DEFERRED/IMMEDIATE
  }

  getReadConnection() {
    // Returns connection configured for read-only
  }

  getWriteConnection() {
    // Returns connection ready for writing
  }
}

import time
import sqlite3
from functools import wraps

def retry_on_busy_snapshot(max_retries=3, base_delay=0.1):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except sqlite3.OperationalError as e:
                    if 'SQLITE_BUSY_SNAPSHOT' in str(e) and attempt < max_retries - 1:
                        delay = base_delay * (2 ** attempt)  # Exponential backoff
                        time.sleep(delay)
                        continue
                    raise
            return func(*args, **kwargs)
        return wrapper
    return decorator

@retry_on_busy_snapshot(max_retries=3)
def update_with_retry(user_id, data):
    with sqlite3.connect('database.db') as conn:
        conn.execute('BEGIN')
        conn.execute('UPDATE users SET data = ? WHERE id = ?', (data, user_id))
        conn.commit()

async function executeWithRetry(operation, maxRetries = 3) {
  for (let i = 0; i < maxRetries; i++) {
    try {
      return await operation();
    } catch (error) {
      if (error.code === 'SQLITE_BUSY_SNAPSHOT' && i < maxRetries - 1) {
        const delay = 100 * Math.pow(2, i); // 100ms, 200ms, 400ms
        await new Promise(resolve => setTimeout(resolve, delay));
        continue;
      }
      throw error;
    }
  }
}

// Usage
await executeWithRetry(async () => {
  return db.transaction(() => {
    db.prepare('UPDATE users SET visits = visits + 1 WHERE id = ?').run(userId);
  });
});

public <T> T retryOnBusySnapshot(Callable<T> operation, int maxRetries) throws Exception {
    for (int i = 0; i < maxRetries; i++) {
        try {
            return operation.call();
        } catch (SQLException e) {
            if (e.getMessage().contains("SQLITE_BUSY_SNAPSHOT") && i < maxRetries - 1) {
                Thread.sleep(100 * (int)Math.pow(2, i)); // Exponential backoff
                continue;
            }
            throw e;
        }
    }
    return operation.call();
}

-- SQLite doesn't have a built-in way to see all transactions,
-- but you can monitor through application logging

-- Check database locks
SELECT * FROM pragma_lock_status;

-- Check WAL mode status
PRAGMA wal_checkpoint(TRUNCATE);

import logging
import sqlite3
import threading

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

class LoggingConnection:
    def __init__(self, db_path):
        self.conn = sqlite3.connect(db_path)
        self.thread_id = threading.get_ident()

    def execute(self, sql, params=()):
        logger.debug(f"[Thread {self.thread_id}] Executing: {sql}")
        try:
            return self.conn.execute(sql, params)
        except sqlite3.OperationalError as e:
            logger.error(f"[Thread {self.thread_id}] Error: {e}")
            raise

    def commit(self):
        logger.debug(f"[Thread {self.thread_id}] Committing transaction")
        self.conn.commit()

    def close(self):
        logger.debug(f"[Thread {self.thread_id}] Closing connection")
        self.conn.close()

import sqlite3

def trace_callback(statement):
    print(f"SQL: {statement}")

conn = sqlite3.connect('database.db')
conn.set_trace_callback(trace_callback)

# On Linux/macOS, monitor file locks
lsof database.db

# Check for lock files
ls -la database.db-*

# Monitor with strace (advanced)
strace -f -e trace=file python your_script.py