Transacciones reactivas en Kotlin: @Transactional, TransactionalOperator y el contexto de corrutinas

Las transacciones clasicas de Spring estan ligadas a un hilo. El framework guarda la conexion JDBC en un `ThreadLocal`, y mientras tu codigo permanezca en ese hilo cada llamada al repositorio se une silenciosamente a la misma transaccion. Spring reactivo descarta ese modelo. Con R2DBC no existe ninguna afinidad de hilo: una sola peticion puede saltar entre muchos hilos de trabajo a medida que se suspende y reanuda, asi que un `ThreadLocal` seria inutil. En su lugar, la transaccion activa se transporta dentro del `Context` de Reactor (o, para corrutinas, el `CoroutineContext`) que viaja con la suscripcion reactiva. Entender este unico cambio es la clave de todo lo demas en esta leccion.

Spring tiende ese puente por ti a traves de `org.springframework.transaction.reactive.TransactionContext`, y la buena noticia es que `@Transactional` sigue funcionando en funciones `suspend`. Cuando anotas una funcion suspend (o una que devuelve `Mono`/`Flux`), el `TransactionInterceptor` de Spring abre una transaccion reactiva, propaga su contexto por la corrutina, y confirma cuando el publicador devuelto se completa con exito o revierte cuando falla. El requisito crucial es que el bean debe ser un componente gestionado por Spring y la llamada debe venir desde fuera del bean, porque el interceptor vive en un proxy. La auto-invocacion (llamar a otro metodo `@Transactional` via `this`) evita el proxy y se ejecuta silenciosamente sin ninguna transaccion.

Para un control programatico mas fino recurres a `TransactionalOperator`. Envuelve un fragmento de trabajo reactivo para que todo lo que se suscriba dentro comparta una transaccion, y expone una extension amigable con corrutinas, `executeAndAwait { ... }`, que te permite escribir codigo suspendido lineal. Es ideal cuando solo quieres que parte de un metodo sea transaccional, cuando necesitas confirmar de forma condicional, o cuando las anotaciones resultan demasiado gruesas. Como el operador gestiona el limite de forma explicita, se compone limpiamente con `coroutineScope` y la concurrencia estructurada, y hace que la region transaccional sea evidente en el punto de llamada en vez de oculta tras una anotacion.

La propagacion se comporta como su pariente bloqueante pero con un giro reactivo. `REQUIRED` (el predeterminado) se une a una transaccion existente si el contexto ya lleva una, o inicia una nueva si no. `REQUIRES_NEW` suspende la transaccion externa y se ejecuta en una totalmente nueva, lo cual es perfecto para una escritura de auditoria que debe persistir aunque la operacion principal revierta. Ten en cuenta que el gestor de transacciones reactivo de R2DBC no soporta todos los modos de propagacion que soporta JPA (por ejemplo, el comportamiento de savepoint `NESTED` no esta soportado), asi que revisa tu driver. El modelo mental que debes conservar es: la propagacion decide si se crea un nuevo `TransactionContext` o se reutiliza uno existente en el contexto del suscriptor.

Ahora el peligro principal: bloquear dentro de una transaccion reactiva. Las tuberias de Reactor y corrutinas se ejecutan sobre un grupo pequeno y fijo de hilos de event-loop no bloqueantes. Si llamas a un repositorio JDBC bloqueante, a `Thread.sleep`, a un `RestTemplate` sincrono, o a cualquier `.block()`/`runBlocking` dentro de una cadena reactiva transaccional, paralizas uno de esos preciados hilos. Bajo carga esto mata de hambre al event-loop, la latencia se dispara, el rendimiento colapsa y las peticiones expiran. Peor aun, el trabajo bloqueante no forma parte del contexto de la transaccion reactiva, asi que una llamada JDBC bloqueante se ejecuta en su propia conexion con su propia transaccion (o ninguna). Tu rollback nunca la deshara, rompiendo silenciosamente la atomicidad.

La solucion es mantener todo el camino transaccional sin bloqueo. Usa repositorios R2DBC, el `DatabaseClient` reactivo y `WebClient` en lugar de sus equivalentes bloqueantes, y nunca llames a `.awaitSingle()` sobre una fuente bloqueante. Si realmente debes invocar codigo bloqueante heredado, llevalo fuera del event-loop con `withContext(Dispatchers.IO)` para no matar de hambre al planificador reactivo, pero entiende claramente que el trabajo bloqueante se ejecuta entonces fuera de la transaccion reactiva y no se revertira con ella. La unica forma de mantener algo dentro de la transaccion es realizarlo a traves de un cliente reactivo, consciente de R2DBC, que participe en el mismo `TransactionContext`.

La semantica del rollback merece una nota. En el mundo reactivo una transaccion revierte cuando el publicador o la funcion suspend devuelta termina con un error. Eso significa que debes dejar que las excepciones se propaguen; si te tragas un error con un `try/catch` y devuelves normalmente, Spring ve exito y confirma. Igualmente, la cancelacion importa: si el scope de la corrutina se cancela a mitad de la transaccion, la transaccion reactiva se revierte, lo cual suele ser lo deseado pero puede sorprender a equipos que migran desde codigo bloqueante donde la cancelacion era rara.

Una lista practica cierra el circulo. Pon `@Transactional` en un metodo publico de un bean de Spring y llamalo desde otro bean; prefiere R2DBC de extremo a extremo; usa `TransactionalOperator.executeAndAwait` cuando necesites limites explicitos; recurre a `REQUIRES_NEW` para efectos secundarios que deben sobrevivir como los registros de auditoria; y audita tu grafo de dependencias en busca de cualquier driver bloqueante escondido tras una interfaz reactiva. Si recuerdas solo una frase, que sea esta: en Spring reactivo la transaccion vive en el contexto que fluye con tus corrutinas, y en el momento en que bloqueas o sales de ese contexto, la garantia desaparece.

import org.springframework.stereotype.Service
import org.springframework.transaction.annotation.Propagation
import org.springframework.transaction.annotation.Transactional

@Service
class OrderService(
    private val orders: OrderRepository,      // R2DBC CoroutineCrudRepository
    private val auditService: AuditService,
) {
    // The interceptor opens a reactive tx, propagates it through the
    // coroutine context, and commits when the function returns normally.
    @Transactional
    suspend fun placeOrder(order: Order): Order {
        val saved = orders.save(order)            // joins the same tx
        auditService.record("order_placed", saved.id) // see REQUIRES_NEW below
        if (saved.total < 0) error("invalid total") // throwing -> rollback
        return saved
    }
}

@Service
class AuditService(private val audit: AuditRepository) {
    // Runs in its own brand-new tx, so the audit row survives even if
    // placeOrder later rolls back. Must be called from another bean.
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    suspend fun record(event: String, id: Long?) {
        audit.save(AuditEntry(event = event, refId = id))
    }
}

import org.springframework.stereotype.Service
import org.springframework.transaction.ReactiveTransactionManager
import org.springframework.transaction.reactive.TransactionalOperator
import org.springframework.transaction.reactive.executeAndAwait

@Service
class TransferService(
    private val accounts: AccountRepository,   // R2DBC, fully reactive
    rtm: ReactiveTransactionManager,           // auto-configured by Spring Boot
) {
    // Spring Boot auto-configures a ReactiveTransactionManager for R2DBC,
    // but NOT a TransactionalOperator — build one from the manager.
    private val tx = TransactionalOperator.create(rtm)

    suspend fun transfer(fromId: Long, toId: Long, amount: Long) =
        // Everything inside shares ONE reactive transaction. If the block
        // throws (or the coroutine is cancelled), the whole thing rolls back.
        tx.executeAndAwait {
            val from = accounts.findById(fromId) ?: error("no source account")
            val to = accounts.findById(toId) ?: error("no target account")
            require(from.balance >= amount) { "insufficient funds" }
            accounts.save(from.copy(balance = from.balance - amount))
            accounts.save(to.copy(balance = to.balance + amount))
        }
}

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext

// ANTI-PATTERN: a blocking call on a reactive event-loop thread.
// It starves the scheduler AND runs outside the reactive transaction,
// so legacyJdbc.write(...) will NOT be rolled back with the tx.
@Transactional
suspend fun broken(order: Order) {
    orders.save(order)              // reactive, in the tx
    legacyJdbc.write(order)         // BLOCKS the event loop, escapes the tx
}

// If you must call blocking code, at least move it off the event loop.
// Still outside the reactive tx — accept that and design around it.
@Transactional
suspend fun lessBad(order: Order) {
    orders.save(order)
    withContext(Dispatchers.IO) { legacyJdbc.write(order) } // no longer starves
}

import kotlinx.coroutines.delay
import kotlinx.coroutines.runBlocking

// Pure stdlib + coroutines demo (no Spring/DB): a non-blocking 'delay'
// suspends without holding a thread, while a blocking sleep would not.
fun main() = runBlocking {
    println("start")
    delay(100)               // suspends, frees the thread (non-blocking)
    // Thread.sleep(100)     // would block the thread — the reactive sin
    println("resumed without parking a thread")
}