Transacciones

Concepto de transacciones

Las transacciones en Valkey garantizan:

• Atomicidad: Todos los comandos se ejecutan o ninguno
• Aislamiento: Ningún otro comando se ejecuta durante la transacción
• No rollback: Si un comando falla, los demás continúan

MULTI/EXEC básico

# Iniciar transacción
MULTI
OK

# Comandos encolados
SET cuenta:1 1000
QUEUED
SET cuenta:2 2000
QUEUED
INCR contador
QUEUED

# Ejecutar todo
EXEC
# 1) OK
# 2) OK
# 3) (integer) 1

Cancelar transacción

MULTI
SET clave "valor"
DISCARD  # Cancela todo
OK

Transferencia entre cuentas

# Transferir 100 de cuenta:1 a cuenta:2
MULTI
DECRBY cuenta:1 100
INCRBY cuenta:2 100
EXEC

WATCH - Optimistic Locking

WATCH permite detectar cambios concurrentes:

# Terminal 1
WATCH saldo
GET saldo
# "1000"

MULTI
SET saldo 900
EXEC
# nil  <- Si otro cliente modificó saldo, falla

# Terminal 2 (mientras Terminal 1 está en MULTI)
SET saldo 500  # Modifica el valor watched

Patrón Check-and-Set (CAS)

WATCH producto:stock
GET producto:stock
# "10"

# Verificar en aplicación que stock > 0

MULTI
DECR producto:stock
RPUSH ordenes '{"producto":1,"cantidad":1}'
EXEC
# Si stock cambió, EXEC retorna nil y reintentamos

Implementación en Python

Transacción simple

import valkey

r = valkey.Valkey(host='localhost', port=6379)

# Usando pipeline con transaction=True
pipe = r.pipeline()
pipe.set('clave1', 'valor1')
pipe.set('clave2', 'valor2')
pipe.incr('contador')
resultados = pipe.execute()

print(resultados)  # [True, True, 1]

Con WATCH

def transferir(origen, destino, monto):
    with r.pipeline() as pipe:
        while True:
            try:
                # Watch las cuentas
                pipe.watch(origen, destino)

                # Leer saldos
                saldo_origen = int(pipe.get(origen) or 0)
                saldo_destino = int(pipe.get(destino) or 0)

                if saldo_origen < monto:
                    pipe.unwatch()
                    raise ValueError("Saldo insuficiente")

                # Iniciar transacción
                pipe.multi()
                pipe.set(origen, saldo_origen - monto)
                pipe.set(destino, saldo_destino + monto)
                pipe.execute()

                return True

            except valkey.WatchError:
                # Alguien modificó las cuentas, reintentar
                continue

# Uso
transferir('cuenta:1', 'cuenta:2', 100)

Reintentos con decorador

from functools import wraps

def transaccion_con_retry(max_retries=3):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for intento in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except valkey.WatchError:
                    if intento == max_retries - 1:
                        raise
                    continue
        return wrapper
    return decorator

@transaccion_con_retry(max_retries=5)
def incrementar_si_menor(clave, limite):
    with r.pipeline() as pipe:
        pipe.watch(clave)
        valor = int(pipe.get(clave) or 0)

        if valor >= limite:
            pipe.unwatch()
            return False

        pipe.multi()
        pipe.incr(clave)
        pipe.execute()
        return True

Implementación en Node.js

import { createClient } from 'valkey';

const client = createClient();
await client.connect();

// Transacción simple
const results = await client.multi()
    .set('clave1', 'valor1')
    .set('clave2', 'valor2')
    .incr('contador')
    .exec();

console.log(results);

// Con WATCH
async function transferir(origen, destino, monto) {
    const maxRetries = 3;

    for (let i = 0; i < maxRetries; i++) {
        try {
            await client.watch([origen, destino]);

            const [saldoOrigen, saldoDestino] = await Promise.all([
                client.get(origen),
                client.get(destino)
            ]);

            if (parseInt(saldoOrigen) < monto) {
                await client.unwatch();
                throw new Error('Saldo insuficiente');
            }

            const results = await client.multi()
                .set(origen, parseInt(saldoOrigen) - monto)
                .set(destino, parseInt(saldoDestino) + monto)
                .exec();

            if (results) return true;

        } catch (err) {
            if (err.message !== 'EXECABORT') throw err;
        }
    }

    throw new Error('Transacción fallida después de reintentos');
}

Errores en transacciones

Error de sintaxis (antes de EXEC)

MULTI
SET clave "valor"
SETT clave2 "valor2"  # Error de sintaxis
# (error) ERR unknown command 'SETT'
EXEC
# (error) EXECABORT Transaction discarded because of previous errors

Error de ejecución (durante EXEC)

SET clave "no_es_numero"

MULTI
INCR clave           # Fallará en ejecución
SET otra "valor"
EXEC
# 1) (error) ERR value is not an integer or out of range
# 2) OK  <- Este sí se ejecuta

Pipeline vs Transaction

Característica	Pipeline	Transaction
Atomicidad	No	Sí
Rendimiento	Alto	Alto
Comandos entrecruzados	Posible	No
Uso de WATCH	No	Sí

# Pipeline sin transacción (más rápido, sin atomicidad)
pipe = r.pipeline(transaction=False)
pipe.set('a', 1)
pipe.set('b', 2)
pipe.execute()

Ejercicios

1. Implementa un contador atómico con límite máximo
2. Crea una transferencia segura entre cuentas
3. Implementa un sistema de reservas con WATCH

Resumen

• MULTI inicia transacción, EXEC la ejecuta
• DISCARD cancela la transacción
• WATCH detecta cambios concurrentes (optimistic locking)
• Las transacciones son atómicas pero no hacen rollback
• Usar reintentos cuando WATCH detecta conflictos

Navegación