Bloqueo optimista en bases de datos

Visión General

Yo uso optimistic locking para prevenir actualizaciones perdidas en entornos concurrentes. Verifico si un registro fue modificado desde su última lectura. Cada fila lleva un número de versión o timestamp. Al actualizar, incluyo la versión original en la cláusula WHERE; si la versión cambió, la actualización falla y reintento o reporto un conflicto. Esto evita el costo de mantener bloqueos de base de datos durante el tiempo de pensamiento del usuario.

Esta implementación proporciona optimistic locking con versionado entero en PostgreSQL y MySQL, además de JPA/Hibernate. Recursos relacionados: Migraciones de Base de Datos de Forma Segura, Gestionar Migraciones de Base de Datos de Forma Segura y Crear y usar vistas y vistas materializadas.

Cuándo Usar

Uso este recurso cuando:

• Varios usuarios, jobs en segundo plano o microservicios a menudo intentan actualizar la misma fila al mismo tiempo. Consulta Database Transactions para patrones ACID.
• Quiero evitar bloqueos pesimistas que dañan throughput y pueden causar deadlocks
• Mi aplicación tiene un patrón de lectura-modificación-escritura con gaps entre lectura y escritura
• Necesito detección de conflictos en APIs REST, apps offline-first o sistemas distribuidos

No lo uso cuando:

• La contención es tan alta que los reintentos se vuelven costosos o impracticables. Para esos casos, prefiero bloqueos pesimistas o operaciones atómicas como SELECT FOR UPDATE.
• Puedo rediseñar el flujo para evitar el patrón lectura-modificación-escritura, por ejemplo apendizando eventos o usando CRDTs.
• Espero que el mismo registro se actualice muchas veces por segundo desde distintas fuentes. A veces un bloqueo pesimista o una cola son más simples.
• Mi base de datos ya soporta aislamiento serializable (p. ej., PostgreSQL SERIALIZABLE) y la carga tolera su overhead.

Solución

Python

import psycopg2
from psycopg2.extras import RealDictCursor

def update_user_email(conn, user_id: int, new_email: str, expected_version: int):
    with conn.cursor(cursor_factory=RealDictCursor) as cur:
        cur.execute("""
            UPDATE users
            SET email = %s, version = version + 1
            WHERE id = %s AND version = %s
            RETURNING id, version;
        """, (new_email, user_id, expected_version))

        updated = cur.fetchone()
        if not updated:
            raise ValueError(
                f"Conflicto: el usuario {user_id} fue modificado por otra transacción. "
                "Por favor refresca y reintenta."
            )
        conn.commit()
        return updated

# Uso
try:
    result = update_user_email(conn, user_id=42, new_email="nuevo@example.com", expected_version=3)
    print(f"Actualizado a versión {result['version']}")
except ValueError as e:
    print(e)  # Disparar lógica de reintento en la capa API

JavaScript

const { Pool } = require('pg');
const pool = new Pool({ /* config */ });

async function updateProductPrice(productId, newPrice, expectedVersion) {
  const result = await pool.query(
    `UPDATE products
     SET price = $1, version = version + 1, updated_at = NOW()
     WHERE id = $2 AND version = $3
     RETURNING id, version;`,
    [newPrice, productId, expectedVersion]
  );

  if (result.rowCount === 0) {
    const current = await pool.query('SELECT version FROM products WHERE id = $1', [productId]);
    throw new Error(
      `Conflicto de versión: esperada ${expectedVersion}, encontrada ${current.rows[0]?.version}. Por favor reintenta.`
    );
  }

  return result.rows[0];
}

// Ruta Express con reintento
app.put('/products/:id', async (req, res) => {
  try {
    const product = await updateProductPrice(req.params.id, req.body.price, req.body.version);
    res.json(product);
  } catch (err) {
    res.status(409).json({ error: err.message });
  }
});

Java

// JPA / Hibernate con @Version
import jakarta.persistence.*;

@Entity
public class Product {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    private String name;
    private BigDecimal price;

    @Version
    private Integer version;  // Auto-incrementado por Hibernate en cada flush

    // Getters y setters...
}

// Capa de servicio
@Service
@Transactional
public class ProductService {
    @Autowired
    private ProductRepository repo;

    public Product updatePrice(Long id, BigDecimal newPrice) {
        Product product = repo.findById(id)
            .orElseThrow(() -> new EntityNotFoundException("Producto no encontrado"));
        product.setPrice(newPrice);
        return repo.save(product);  // Versión verificada automáticamente en flush
    }
}

// Capturar la excepción de optimistic lock
@ExceptionHandler(OptimisticLockingFailureException.class)
public ResponseEntity<Map<String, String>> handleConflict(OptimisticLockingFailureException ex) {
    return ResponseEntity.status(HttpStatus.CONFLICT)
        .body(Map.of("error", "Recurso modificado por otro usuario. Por favor refresca y reintenta."));
}

Explicación

Yo pienso en optimistic locking como una apuesta a que los conflictos serán raros. La base de datos no bloquea la fila mientras leo. En su lugar, la actualización es condicional:

UPDATE table SET ... WHERE id = ? AND version = ?

Si rowsAffected == 0, la versión cambió entre lectura y escritura. Yo manejo el conflicto: reintento con datos frescos, devuelvo HTTP 409, o fusiono cambios.

Compromisos:

• Optimista: las lecturas quedan libres de bloqueos y el sistema escala, pero tengo que manejar conflictos y reintentar.
• Pesimista: SELECT FOR UPDATE bloquea la fila de inmediato; la lógica es más simple, pero serializa el acceso y puede generar deadlocks.

Para más patrones de concurrencia, consulta Concurrent Data Structures.

Variantes

Tecnología	Enfoque	Notas
Versión entera	Columna version incrementada en cada actualización	Más común; funciona en todas las bases de datos relacionales
Timestamp	Columna updated_at comparada al escribir	Propenso a problemas de reloj; usa timestamps de base de datos, no de aplicación
Checksum / hash	Hash del contenido de fila almacenado y comparado	Detecta cualquier cambio, incluso si el versionado fue evadido
JPA @Version	Versionado entero automático	Hibernate maneja incremento y detección de conflictos transparentemente
DynamoDB	Escrituras condicionales con Expected	Sin versionado nativo; usa attribute_exists o comparaciones de valores
MongoDB	findAndModify con criterios de consulta	Incluye versión en filtro; reintenta si el documento fue modificado

Lo que funciona

1. Siempre devuelvo la versión actual al cliente después de cada lectura para que pueda enviarla en la actualización.
2. Implemento reintento con backoff exponencial (1–3 intentos) para conflictos transitorios en procesos automatizados.
3. Uso version entero sobre timestamps; en producción he visto relojes desfasarse y causar falsos positivos entre nodos.
4. Mantengo las transacciones cortas; el gap entre lectura y escritura es la ventana de vulnerabilidad.
5. Registro conflictos de versión a nivel INFO para monitorear hotspots de contención sin alarmar en cada reintento.

Errores Comunes

1. No exponer la versión a consumidores de API — los clientes no pueden enviarla si nunca la recibieron. He depurado esto en una API donde el frontend perdía silenciosamente cada segunda actualización.
2. Bucles de reintento infinitos — siempre limito reintentos y expongo conflictos persistentes al usuario.
3. Actualizar la versión en código de aplicación — dejo que la base de datos u ORM la incremente atómicamente.
4. Usar bloqueo pesimista para todo — mata el throughput; reservo FOR UPDATE para verdaderos escenarios de inventario o banca. Consulta Locks and Mutexes para patrones de bloqueo.
5. Ignorar el conflicto en UI — los usuarios necesitan retroalimentación clara de que sus datos están obsoletos y deben refrescarse.

Ejemplos de Implementación

Lógica de reintento con exponential backoff

import random
import time
from functools import wraps

def retry_on_conflict(max_retries=3, base_delay=0.05):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except ValueError as e:
                    if "Conflict" not in str(e):
                        raise
                    if attempt == max_retries - 1:
                        raise
                    delay = base_delay * (2 ** attempt) + random.uniform(0, 0.05)
                    time.sleep(delay)
            return None
        return wrapper
    return decorator

@retry_on_conflict(max_retries=3)
def update_user_with_retry(conn, user_id, new_email, expected_version):
    return update_user_email(conn, user_id, new_email, expected_version)

async function withRetry(fn, maxRetries = 3, baseDelay = 50) {
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      return await fn();
    } catch (err) {
      if (!err.message.includes('Version conflict') || attempt === maxRetries - 1) {
        throw err;
      }
      const delay = baseDelay * (2 ** attempt) + Math.random() * 50;
      await new Promise(r => setTimeout(r, delay));
    }
  }
}

// Uso con refresh automático de versión
async function updateProductWithRetry(productId, updateFn) {
  let product = await getProduct(productId);
  for (let attempt = 0; attempt < 3; attempt++) {
    const updated = updateFn(product);
    try {
      return await pool.query(
        'UPDATE products SET price = $1, version = version + 1 WHERE id = $2 AND version = $3 RETURNING *',
        [updated.price, productId, product.version]
      );
    } catch (err) {
      if (attempt === 2) throw err;
      product = await getProduct(productId);
    }
  }
}

Bloqueo optimista en MongoDB con findAndModify

const { MongoClient } = require('mongodb');
const client = new MongoClient(process.env.MONGO_URI);

async function updateProductOptimistic(db, productId, newPrice, expectedVersion) {
  const result = await db.collection('products').findOneAndUpdate(
    { _id: productId, version: expectedVersion },
    {
      $set: { price: newPrice },
      $inc: { version: 1 },
    },
    { returnDocument: 'after' }
  );

  if (!result) {
    const current = await db.collection('products').findOne({ _id: productId });
    throw new Error(
      `Conflicto de versión: esperada ${expectedVersion}, encontrada ${current?.version}. Reintenta.`
    );
  }

  return result;
}

// Plugin de Mongoose para versionado automático
const optimisticLockPlugin = (schema) => {
  schema.add({ version: { type: Number, default: 0 } });

  schema.pre('findOneAndUpdate', function () {
    const filter = this.getFilter();
    const update = this.getUpdate();

    if (filter.version !== undefined && update.$inc) {
      update.$inc.version = (update.$inc.version || 0) + 1;
    } else if (filter.version !== undefined) {
      this.setUpdate({ ...update, $inc: { version: 1 } });
    }
  });
};

productSchema.plugin(optimisticLockPlugin);

Escrituras condicionales en DynamoDB

import boto3

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('products')

def update_price_optimistic(product_id, new_price, expected_version):
    response = table.update_item(
        Key={'product_id': product_id},
        UpdateExpression='SET price = :p, version = :new_v',
        ConditionExpression='version = :expected',
        ExpressionAttributeValues={
            ':p': new_price,
            ':new_v': expected_version + 1,
            ':expected': expected_version,
        },
        ReturnValues='ALL_NEW'
    )
    return response

# Manejar fallo de conditional check
from botocore.exceptions import ClientError

try:
    update_price_optimistic('prod-42', 99.99, 3)
except ClientError as e:
    if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
        print("Conflicto de versión: otro proceso modificó este item")

ETag e If-Match para APIs HTTP

// Middleware Express para optimistic locking basado en ETag
const crypto = require('crypto');

function generateETag(resource) {
  const hash = crypto.createHash('md5');
  hash.update(JSON.stringify(resource));
  return `"${hash.digest('hex')}"`;
}

app.put('/products/:id', async (req, res) => {
  const ifMatch = req.headers['if-match'];
  if (!ifMatch) {
    return res.status(428).json({ error: 'Header If-Match requerido' });
  }

  const product = await getProduct(req.params.id);
  const currentETag = generateETag(product);

  if (ifMatch !== currentETag) {
    return res.status(412).json({
      error: 'Precondition failed: el recurso fue modificado',
      currentETag,
    });
  }

  const updated = await updateProduct(req.params.id, req.body);
  res.set('ETag', generateETag(updated));
  res.json(updated);
});

Batch optimistic locking

def batch_update_with_versions(conn, updates):
    """Actualizar múltiples filas con optimistic locking en una sola transacción."""
    results = []
    with conn.cursor() as cur:
        for item in updates:
            cur.execute("""
                UPDATE products
                SET price = %s, version = version + 1
                WHERE id = %s AND version = %s
                RETURNING id, version;
            """, (item['new_price'], item['id'], item['expected_version']))

            updated = cur.fetchone()
            if not updated:
                conn.rollback()
                raise ValueError(
                    f"Conflicto en producto {item['id']}: "
                    f"versión esperada {item['expected_version']}"
                )
            results.append(updated)
    conn.commit()
    return results

# Uso
try:
    results = batch_update_with_versions(conn, [
        {'id': 1, 'new_price': 19.99, 'expected_version': 5},
        {'id': 2, 'new_price': 29.99, 'expected_version': 3},
        {'id': 3, 'new_price': 39.99, 'expected_version': 7},
    ])
except ValueError as e:
    print(f"Batch falló: {e}")
    # Todos los updates se revirtieron, el cliente debe refrescar y reintentar

Estrategias de resolución de conflictos

Un patrón común es hacer merge de campos no superpuestos. Si el cliente cambió el email y el servidor cambió el nombre, puedes conservar ambos. La clave es leer la versión actual, mezclar los cambios y escribir con un nuevo chequeo de versión:

def merge_update(conn, user_id, client_changes, expected_version):
    with conn.cursor() as cur:
        cur.execute("SELECT name, email, version FROM users WHERE id = %s", (user_id,))
        current = cur.fetchone()
        if not current:
            raise ValueError("Usuario no encontrado")

        merged = {
            'name': client_changes.get('name', current['name']),
            'email': client_changes.get('email', current['email']),
        }

        cur.execute("""
            UPDATE users
            SET name = %s, email = %s, version = version + 1
            WHERE id = %s AND version = %s
            RETURNING id, version;
        """, (merged['name'], merged['email'], user_id, current['version']))

        updated = cur.fetchone()
        if not updated:
            raise ValueError("Conflicto: el registro cambió durante el merge. Reintenta.")
        conn.commit()
        return updated

Si los campos se superponen, la decisión es específica del dominio: muestra un diff al usuario, elige un ganador o pide confirmación.

Notas de Producción

1. Indexa la columna version. La cláusula WHERE id = ? AND version = ? necesita un índice en ambas columnas:

CREATE INDEX idx_products_id_version ON products (id, version);

2. Mantén corto el gap read-modify-write. Cuanto más largo el gap, más probables los conflictos. Yo evito llamar APIs externas o hacer cómputo pesado entre read y write.

3. Usa RETURNING para evitar una segunda consulta. Obtengo la versión actualizada en la misma sentencia:

UPDATE products SET price = $1, version = version + 1
WHERE id = $2 AND version = $3
RETURNING id, version;

4. Monitorea tasas de conflicto con pg_stat_database. Yo rastreo deadlocks y conflictos a nivel de base de datos:

SELECT datname, deadlocks, conflicts, temp_files
FROM pg_stat_database
WHERE datname = current_database();

5. Considera SERIALIZABLE isolation en lugar de versionado manual. PostgreSQL SERIALIZABLE maneja conflictos automáticamente usando SSI (Serializable Snapshot Isolation). Lo he encontrado más simple que versionar manualmente en transacciones complejas.

Puntos Clave

• El bloqueo optimista evita bloqueos largos haciendo que cada actualización sea condicional a un número de versión. Por eso lo prefiero para cargas de lectura intensiva.
• Yo uso una columna version entera en lugar de timestamps; la incremento atómicamente en la base de datos u ORM.
• Siempre devuelvo la versión actual en lecturas, y devuelvo un 409 Conflict claro cuando la versión no coincide.
• Mantengo corta la ventana de lectura-modificación-escritura y limito reintentos para evitar avalanchas.
• Cuando la contención es alta, recurre a bloqueos pesimistas, SELECT FOR UPDATE o aislamiento serializable.

Lectura Adicional

• PostgreSQL concurrency control
• MySQL locking reads
• Jakarta Persistence @Version
• Database Transactions para patrones ACID
• Retry Backoff para lógica de reintento
• Locks and Mutexes para patrones de bloqueo pesimista